vector: add optional buffer consumer to ingest from buffer queue (opendata-oss#435)

Adds a `buffer` cargo feature that wires an `opendata-buffer` Consumer into
the vector server. When configured, the server runs a background task that
polls the queue, decodes each entry as a protobuf `WriteRequest` (with a
2-byte version/encoding prefix), and writes through `VectorDb`. The
consumer starts before the HTTP listener and is shut down before the
server exits.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: cadonna

Cargo.lock

@@ -23,6 +23,7 @@ required-features = ["http-server"]
 default = []
 http-server = ["dep:axum", "dep:tower", "dep:clap", "dep:prost", "dep:serde_json", "dep:serde_with"]
 embedded-reader = ["dep:reqwest", "dep:serde_json"]
+buffer = ["http-server", "dep:buffer"]
 bench-internals = []
 avx512 = []
 
@@ -60,6 +61,9 @@ serde_with = { workspace = true, optional = true }
 # Embedded reader dependencies (optional, enabled by embedded-reader feature)
 reqwest = { workspace = true, optional = true, features = ["json"] }
 
+# Buffer consumer dependencies (optional, enabled by buffer feature)
+buffer = { workspace = true, optional = true }
+
 [[bin]]
 name = "gen_sift100k_groundtruth"
 path = "src/bin/gen_sift100k_groundtruth.rs"

vector/Cargo.toml

@@ -59,12 +59,23 @@ async fn main() {
             let vector_config = load_vector_config(&config);
             tracing::info!("Opening vector database with config: {:?}", vector_config);
             let metadata_fields = vector_config.metadata_fields.clone();
+            #[cfg(feature = "buffer")]
+            let buffer_consumer_config = vector_config.buffer_consumer.clone();
 
-            let db = VectorDb::open(vector_config)
-                .await
-                .expect("Failed to open vector database");
+            let db = Arc::new(
+                VectorDb::open(vector_config)
+                    .await
+                    .expect("Failed to open vector database"),
+            );
+
+            let server = VectorServer::new(db, server_config, metadata_fields);
+
+            #[cfg(feature = "buffer")]
+            let server = match buffer_consumer_config {
+                Some(buffer_config) => server.with_buffer_consumer(buffer_config),
+                None => server,
+            };
 
-            let server = VectorServer::new(Arc::new(db), server_config, metadata_fields);
             server.run().await;
         }
         Command::Reader { config } => {

vector/src/bin/server.rs

@@ -7,6 +7,8 @@
 use std::collections::HashMap;
 use std::time::Duration;
 
+#[cfg(feature = "buffer")]
+use common::ObjectStoreConfig;
 use common::StorageConfig;
 use serde::{Deserialize, Serialize};
 
@@ -245,6 +247,12 @@ pub struct Config {
     /// attribute names or type mismatches will fail. If empty, any attribute
     /// names are accepted with types inferred from the first write.
     pub metadata_fields: Vec<MetadataFieldSpec>,
+
+    /// Buffer consumer configuration. When `Some`, the server starts a
+    /// background task that ingests vectors from an `opendata-buffer` queue.
+    #[cfg(feature = "buffer")]
+    #[serde(default)]
+    pub buffer_consumer: Option<BufferConsumerConfig>,
 }
 
 impl Default for Config {
@@ -263,10 +271,68 @@ impl Default for Config {
             chunk_target: 4096,
             query_pruning_factor: None,
             metadata_fields: Vec::new(),
+            #[cfg(feature = "buffer")]
+            buffer_consumer: None,
         }
     }
 }
 
+/// Configuration for the embedded buffer consumer task.
+///
+/// Mirrors [`buffer::ConsumerConfig`], plus a poll interval used by the
+/// vector consumer when the queue is empty.
+#[cfg(feature = "buffer")]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct BufferConsumerConfig {
+    /// Object store where the buffer queue lives. Must match the producer.
+    pub object_store: ObjectStoreConfig,
+
+    /// Path to the queue manifest in object storage. Must match the producer.
+    #[serde(default = "default_buffer_manifest_path")]
+    pub manifest_path: String,
+
+    /// Path prefix for data batch objects. Must match the producer.
+    #[serde(default = "default_buffer_data_path_prefix")]
+    pub data_path_prefix: String,
+
+    /// Poll interval when the queue is empty.
+    #[serde(default = "default_buffer_poll_interval", with = "duration_secs")]
+    pub poll_interval: Duration,
+
+    /// How often the garbage collector runs.
+    #[serde(default = "default_buffer_gc_interval", with = "duration_secs")]
+    pub gc_interval: Duration,
+
+    /// Minimum age before an unreferenced batch file is deleted.
+    #[serde(default = "default_buffer_gc_grace_period", with = "duration_secs")]
+    pub gc_grace_period: Duration,
+}
+
+#[cfg(feature = "buffer")]
+fn default_buffer_manifest_path() -> String {
+    "ingest/manifest".to_string()
+}
+
+#[cfg(feature = "buffer")]
+fn default_buffer_data_path_prefix() -> String {
+    "ingest".to_string()
+}
+
+#[cfg(feature = "buffer")]
+fn default_buffer_poll_interval() -> Duration {
+    Duration::from_millis(100)
+}
+
+#[cfg(feature = "buffer")]
+fn default_buffer_gc_interval() -> Duration {
+    Duration::from_secs(300)
+}
+
+#[cfg(feature = "buffer")]
+fn default_buffer_gc_grace_period() -> Duration {
+    Duration::from_secs(600)
+}
+
 /// Options for search operations.
 #[derive(Debug, Clone, Default)]
 pub struct SearchOptions {