changes to keep

crates/doc/src/combine/memtable.rs

@@ -217,6 +217,12 @@ impl MemTable {
         &self.zz_alloc
     }
 
+    /// Parse a JSON document string into a HeapNode using this MemTable's allocator.
+    pub fn parse_json_str<'s>(&'s self, doc_json: &str) -> serde_json::Result<HeapNode<'s>> {
+        let mut de = serde_json::Deserializer::from_str(doc_json);
+        HeapNode::from_serde(&mut de, self.alloc())
+    }
+
     /// Add the document to the MemTable.
     pub fn add<'s>(&'s self, binding: u32, root: HeapNode<'s>, front: bool) -> Result<(), Error> {
         // Safety: mutable borrow does not escape this function.

crates/doc/src/combine/mod.rs

@@ -192,6 +192,14 @@ impl Iterator for Drainer {
 }
 
 impl Drainer {
+    /// Drain the next document of this Drainer.
+    pub fn drain_next(&mut self) -> Result<Option<DrainedDoc>, Error> {
+        match self {
+            Self::Mem { drainer, .. } => drainer.drain_next(),
+            Self::Spill { drainer } => drainer.drain_next(),
+        }
+    }
+
     /// Map this Drainer into a new and empty Accumulator.
     /// Any un-drained documents are dropped.
     pub fn into_new_accumulator(self) -> Result<Accumulator, Error> {

crates/doc/src/combine/spill.rs

@@ -325,7 +325,7 @@ pub struct SpillDrainer<F: io::Read + io::Seek> {
 unsafe impl<F: io::Read + io::Seek> Send for SpillDrainer<F> {}
 
 impl<F: io::Read + io::Seek> SpillDrainer<F> {
-    fn drain_next(&mut self) -> Result<Option<DrainedDoc>, Error> {
+    pub fn drain_next(&mut self) -> Result<Option<DrainedDoc>, Error> {
         let Some(cmp::Reverse(segment)) = self.heap.pop() else {
             return Ok(None);
         };

crates/doc/src/reduce/mod.rs

@@ -315,11 +315,12 @@ fn compare_lazy<L: AsNode, R: AsNode>(
 /// merge_patch_schema returns a JSON-Schema implementing the RFC-7396 Merge patch algorithm.
 pub fn merge_patch_schema() -> serde_json::Value {
     serde_json::json!({
+        "$id": "flow://merge-patch-schema",
         "oneOf": [
             {
                 "type": "object",
                 "reduce": {"strategy": "merge"},
-                "additionalProperties": {"$ref": "#"}
+                "additionalProperties": {"$ref": "flow://merge-patch-schema"}
             },
             {
                 "type": "null",

crates/flowctl/src/raw/spec.rs

@@ -80,8 +80,7 @@ pub async fn do_spec(
         format!("spec/{}", capture.capture),
     )
     .unary_capture(spec_req, build::CONNECTOR_TIMEOUT)
-    .await
-    .map_err(crate::status_to_anyhow)?
+    .await?
     .spec
     .context("connector didn't send expected Spec response")?;
 

crates/ops/src/lib.rs

@@ -5,9 +5,9 @@ use std::io::Write;
 pub mod decode;
 pub mod tracing;
 
-pub use proto_flow::ops::log::Level as LogLevel;
-pub use proto_flow::ops::Log;
-pub use proto_flow::ops::TaskType;
+// Re-export many types from proto_flow::ops, so that users of this crate
+// don't also have to use that module.
+pub use proto_flow::ops::{log::Level as LogLevel, stats, Log, Meta, ShardRef, Stats, TaskType};
 
 #[derive(Serialize, Deserialize, Clone, Debug)]
 #[serde(rename_all = "camelCase")]
@@ -22,7 +22,7 @@ pub struct Shard {
     r_clock_begin: HexU32,
 }
 
-impl From<Shard> for proto_flow::ops::ShardRef {
+impl From<Shard> for ShardRef {
     fn from(
         Shard {
             kind,

crates/proto-flow/src/runtime.rs

@@ -152,12 +152,58 @@ pub struct Container {
 pub struct CaptureRequestExt {
     #[prost(message, optional, tag = "1")]
     pub labels: ::core::option::Option<super::ops::ShardLabeling>,
+    #[prost(message, optional, tag = "2")]
+    pub open: ::core::option::Option<capture_request_ext::Open>,
+    #[prost(message, optional, tag = "3")]
+    pub start_commit: ::core::option::Option<capture_request_ext::StartCommit>,
+}
+/// Nested message and enum types in `CaptureRequestExt`.
+pub mod capture_request_ext {
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, PartialEq, ::prost::Message)]
+    pub struct Open {
+        /// RocksDB descriptor which should be opened.
+        #[prost(message, optional, tag = "1")]
+        pub rocksdb_descriptor: ::core::option::Option<super::RocksDbDescriptor>,
+    }
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, PartialEq, ::prost::Message)]
+    pub struct StartCommit {
+        /// Flow runtime checkpoint associated with this transaction.
+        #[prost(message, optional, tag = "1")]
+        pub runtime_checkpoint: ::core::option::Option<
+            ::proto_gazette::consumer::Checkpoint,
+        >,
+    }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CaptureResponseExt {
     #[prost(message, optional, tag = "1")]
     pub container: ::core::option::Option<Container>,
+    #[prost(message, optional, tag = "2")]
+    pub captured: ::core::option::Option<capture_response_ext::Captured>,
+    #[prost(message, optional, tag = "4")]
+    pub checkpoint: ::core::option::Option<capture_response_ext::Checkpoint>,
+}
+/// Nested message and enum types in `CaptureResponseExt`.
+pub mod capture_response_ext {
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, PartialEq, ::prost::Message)]
+    pub struct Captured {
+        /// Packed key extracted from the captured document.
+        #[prost(bytes = "bytes", tag = "1")]
+        pub key_packed: ::prost::bytes::Bytes,
+        /// Packed partition values extracted from the captured document.
+        #[prost(bytes = "bytes", tag = "2")]
+        pub partitions_packed: ::prost::bytes::Bytes,
+    }
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, PartialEq, ::prost::Message)]
+    pub struct Checkpoint {
+        #[prost(message, optional, tag = "1")]
+        pub stats: ::core::option::Option<super::super::ops::Stats>,
+    }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
@@ -227,6 +273,18 @@ pub mod derive_response_ext {
 pub struct MaterializeRequestExt {
     #[prost(message, optional, tag = "1")]
     pub labels: ::core::option::Option<super::ops::ShardLabeling>,
+    #[prost(message, optional, tag = "2")]
+    pub open: ::core::option::Option<materialize_request_ext::Open>,
+}
+/// Nested message and enum types in `MaterializeRequestExt`.
+pub mod materialize_request_ext {
+    #[allow(clippy::derive_partial_eq_without_eq)]
+    #[derive(Clone, PartialEq, ::prost::Message)]
+    pub struct Open {
+        /// RocksDB descriptor which should be opened.
+        #[prost(message, optional, tag = "1")]
+        pub rocksdb_descriptor: ::core::option::Option<super::RocksDbDescriptor>,
+    }
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

crates/runtime/Cargo.toml

@@ -15,6 +15,7 @@ coroutines = { path = "../coroutines"  }
 derive-sqlite = { path = "../derive-sqlite" }
 doc = { path = "../doc" }
 extractors = { path = "../extractors" }
+json = { path = "../json" }
 locate-bin = { path = "../locate-bin" }
 models = { path = "../models" }
 ops = { path = "../ops" }

crates/runtime/src/materialize/mod.rs

@@ -130,6 +130,7 @@ where
     request_rx.inspect_ok(move |request| {
         let Ok(MaterializeRequestExt {
             labels: Some(ops::ShardLabeling { log_level, .. }),
+            ..
         }) = request.get_internal()
         else {
             return;