Test timely capability ref PR

Cargo.toml

@@ -75,3 +75,10 @@ rustc-demangle = { opt-level = 3 }
 # it's unlikely we're going to get interactive access to a debugger in
 # production installations, but we still want useful crash reports.
 debug = 1
+
+
+[patch."https://github.com/TimelyDataflow/timely-dataflow"]
+timely = { git = "https://github.com/petrosagg/timely-dataflow", branch = "safe-capabilityref" }
+
+[patch."https://github.com/TimelyDataflow/differential-dataflow"]
+differential-dataflow = { git = "https://github.com/petrosagg/differential-dataflow", branch = "safe-capref" }

deny.toml

@@ -85,6 +85,8 @@ allow-git = [
     "https://github.com/MaterializeInc/rust-prometheus.git",
     "https://github.com/TimelyDataflow/timely-dataflow",
     "https://github.com/TimelyDataflow/differential-dataflow.git",
+    "https://github.com/petrosagg/timely-dataflow",
+    "https://github.com/petrosagg/differential-dataflow.git",
     "https://github.com/fede1024/rust-rdkafka.git",
     # Waiting for a new release that includes the latest dependency bumps.
     "https://github.com/tikv/fail-rs.git",

src/interchange/src/envelopes.rs

@@ -36,7 +36,7 @@ where
     let x: Stream<G, ((Option<Row>, Vec<DiffPair<Row>>), G::Timestamp, Diff)> =
         batches.unary(Pipeline, "combine_at_timestamp", move |_, _| {
             move |input, output| {
-                while let Some((cap, batches)) = input.next() {
+                while let Some((cap, batches)) = input.next().with_cap() {
                     let mut session = output.session(&cap);
                     batches.swap(&mut rows_buf);
                     for batch in rows_buf.drain(..) {

src/persist/src/operators/async_ext.rs

@@ -213,9 +213,7 @@ mod test {
                         // Drop initial capabilities
                         capabilities.clear();
                         let mut output_handle = output.activate();
-                        while let Some((cap, data)) = input_handle.next() {
-                            let cap = cap.retain();
-
+                        while let Some((cap, data)) = input_handle.next().with_cap() {
                             let mut session = output_handle.session(&cap);
                             for item in data.iter().copied() {
                                 tokio::time::sleep(Duration::from_millis(10)).await;

src/persist/src/operators/stream.rs

@@ -686,10 +686,7 @@ where
                 let mut data_output = data_output.activate();
 
                 // Write out everything and forward, keeping the write futures.
-                while let Some((cap, data)) = input.next() {
-                    // TODO(petrosagg): remove this unconditional retain once this is released:
-                    //     https://github.com/TimelyDataflow/timely-dataflow/pull/429
-                    let cap = cap.retain();
+                while let Some((cap, data)) = input.next().with_cap() {
                     data.swap(&mut buffer);
 
                     let mut session = data_output.session(&cap);

src/sql/Cargo.toml

@@ -45,3 +45,6 @@ tokio-postgres = { git = "https://github.com/MaterializeInc/rust-postgres", bran
 uncased = "0.9.6"
 url = "2.2.2"
 uuid = { version = "0.8.2", features = ["serde", "v4"] }
+
+[dev-dependencies]
+datadriven = "0.6.0"