From 5d9d046d728467c44d5b807f6e4df376c0216f88 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Fri, 23 Oct 2026 23:55:47 -0400
Subject: [PATCH 1/3] DDIR: every import is read as of the time its reader
 starts

A program installed after a published trace had compacted read the trace's
history as stored. Before the compaction frontier a batch can hold a
retraction advanced to the frontier beside the insertion it retracts, which
is a count no time had. An iterative scope seeded from such an import need
not converge, and `tick` hung: 3 of 620 differential test seeds with late
installs.

`Server::import_as_of()` is the epoch before the current one, which is also
the frontier `tick` compacts every published trace to. Every import reads as
of it:
- the corgi arrangement `Arr` carries `since`, and a join or reduce of an
  import reads through DD's `BackendFrontier` (a join of two imports reads as
  of the join of their times);
- a collection read from an import has its copied time lanes advanced;
- programs, bindings, snapshots and tails all import through the same path.

The half join reads the arrangement as stored: with the stream at or beyond
the as-of time and published times at the root, the rows it meets and the
times it produces are the same. An arrangement the program built is read as
it is, and an import as of the minimum constructs no wrapper.

The differential tests run with late installs (`LATE_INSTALLS`), and
`reinstall` with them. `late_install_regressions` runs the three seeds that
hung (40000075, 40000160, 40000519) under `split`, `bind` and `reinstall`;
without the as-of reads it does not finish. `server_late_import.rs` reads
imports as containers, as rows, and as arrangements read by a join, a reduce
and a loop, installed late.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o
---
 interactive/server/src/loop_.rs         |   2 +-
 interactive/src/backend/corgi.rs        |  98 +++++++++++----
 interactive/src/server/bind.rs          |   7 +-
 interactive/src/server/install.rs       |  14 ++-
 interactive/src/server/mod.rs           |  20 +++-
 interactive/src/server/read.rs          |   7 +-
 interactive/tests/difftest.rs           |  26 +++-
 interactive/tests/server_late_import.rs | 151 ++++++++++++++++++++++++
 8 files changed, 285 insertions(+), 40 deletions(-)
 create mode 100644 interactive/tests/server_late_import.rs

diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
index 12f39cbf1..fdca2291c 100644
--- a/interactive/server/src/loop_.rs
+++ b/interactive/server/src/loop_.rs
@@ -477,7 +477,7 @@ fn start_tail(
     let tag = reqid.to_string();
     let mut probe = ProbeHandle::new();
     let shutdown = interactive::server::host_dataflow(worker, |_, scope| {
-        let (containers, shutdown) = trace.import_containers(scope.clone(), "TailImport");
+        let (containers, shutdown) = trace.import_containers(scope, "TailImport", server.import_as_of());
         let pending = Arc::new(AtomicUsize::new(0));
         let (mut upper, mut dropped) = (None::<OuterTime>, false);
         containers
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 0d797aa44..5a38936c5 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -10,7 +10,7 @@
 use timely::dataflow::Scope;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::generic::Operator;
-use timely::progress::Antichain;
+use timely::progress::{Antichain, Timestamp};
 
 use differential_dataflow::AsCollection;
 use differential_dataflow::Collection;
@@ -34,7 +34,7 @@ use crate::runtime::half_join::{CorgiHalfJoinBackend, HalfJoinBatcher};
 use crate::runtime::join::CorgiJoinBackend;
 use differential_dogs3::operators::half_join::half_join_with_tactic;
 use crate::runtime::reduce::CorgiReduceBackend;
-use differential_dataflow::operators::int_proxy::{ProxyJoinTactic, ProxyReduceTactic};
+use differential_dataflow::operators::int_proxy::{BackendFrontier, ProxyJoinTactic, ProxyReduceTactic};
 use differential_dogs3::operators::half_join::proxy::ProxyHalfJoinTactic;
 use crate::lambda::{compile_flatmap, compile_join_function, compile_predicate, compile_projection, compile_scalar};
 use corgi::{Graph, NumOp, Shape};
@@ -230,7 +230,29 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &[Plan]) -> CC {
 
 /// The corgi backend's arrangement: one a program built, or a published export it imported.
 /// Both are the same trace type at the same time, so every operator reads either in place.
-pub(crate) type Arr<'scope> = Arranged<'scope, TraceAgent<CTrace>>;
+///
+/// An import is read as of `since`, the time its reader started: a published trace has compacted
+/// to that time, from which on its accumulations are correct, and before which its batches can
+/// hold counts no time had. Every reader advances the times it draws to `since`: a collection as
+/// it flattens the chunks, a join or reduce through DD's `BackendFrontier`. An arrangement the
+/// program built is read as it is, which `since` at the minimum says.
+///
+/// One time, not a frontier: the server compacts every published trace by the one-element
+/// frontier `{(epoch - 1, [])}` and a reader starts at that same element, so the frontier a
+/// reader reads as of is `{since}`, which is beyond the frontier its trace compacted by, as DD's
+/// `BackendFrontier` requires.
+#[derive(Clone)]
+pub(crate) struct Arr<'scope> {
+    pub(crate) arranged: Arranged<'scope, TraceAgent<CTrace>>,
+    pub(crate) since: Time,
+}
+
+impl<'scope> Arr<'scope> {
+    /// An arrangement the program built, read as it is.
+    fn built(arranged: Arranged<'scope, TraceAgent<CTrace>>) -> Self {
+        Arr { arranged, since: Time::minimum() }
+    }
+}
 
 /// The time each coordinate of `t` less one, at least zero: the half join's compaction of its
 /// arrangement. Every later stream time is at or beyond some such frontier element, and advancing
@@ -241,16 +263,20 @@ fn step_back(t: &Time, out: &mut Antichain<Time>) {
     out.insert(timely::order::Product::new(t.outer.saturating_sub(1), inner));
 }
 
-/// The updates of a stream of columnar batches, as corgi containers.
+/// The updates of a stream of columnar batches, as corgi containers, read as of `since`.
 ///
 /// Each chunk already IS a columnar container: its key/val columns clone by Arc bump,
 /// so a chunk becomes a `CorgiContainer` for the price of copying its time lanes and
 /// its diffs. One container per chunk — no concatenation, no gather, no
-/// columns→rows→columns round-trip.
+/// columns→rows→columns round-trip. The copied time lanes are advanced to `since` in
+/// place, one lane-wise max; a container's rows may then coincide, which the arrange
+/// that reads them consolidates.
 fn chunk_containers<'s, T: ColTime>(
     stream: timely::dataflow::Stream<'s, T, Vec<Span<T, std::rc::Rc<ChunkBatch<CorgiChunk<T, Diff>>>>>>,
     name: &str,
+    since: T,
 ) -> Collection<'s, T, CorgiContainer<T, Diff>> {
+    let since = (since != T::minimum()).then(|| Antichain::from_elem(since));
     stream
         .unary(Pipeline, name, move |_, _| {
             move |input, output| {
@@ -259,12 +285,16 @@ fn chunk_containers<'s, T: ColTime>(
                     for batch in data.iter() {
                         let Some(payload) = batch.inner.clone() else { continue };
                         for ch in payload.chunks.iter().filter(|c| c.len() > 0) {
+                            let mut times = ch.times().clone();
+                            if let Some(since) = &since {
+                                times.advance_by(since.borrow(), times.len());
+                            }
                             let mut c = CorgiContainer {
                                 // Drop the arrangement's leading identifier lane: edges carry
                                 // the key the program wrote, so `$0` indexes what it always did.
                                 keys: recover_key(ch.keys()),
                                 vals: ch.vals().clone(),
-                                times: ch.times().clone(),
+                                times,
                                 diffs: ch.diffs().to_vec(),
                             };
                             session.give_container(&mut c);
@@ -316,16 +346,16 @@ impl Backend for CorgiBackend {
         // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s
         // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher +
         // builder. No columns→rows→columns round-trip at the arrangement boundary.
-        arrange_core::<_, CC, _, CTrace>(
+        Arr::built(arrange_core::<_, CC, _, CTrace>(
             c.inner,
             CorgiPact,
             "CorgiArrange",
             ChunkBatcher::<CorgiChunker<Time, Diff>, _>::new,
-        )
+        ))
     }
 
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, CC> {
-        chunk_containers(a.stream, "CorgiAsCollection")
+        chunk_containers(a.arranged.stream, "CorgiAsCollection", a.since)
     }
 
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Lambda, ((k, v0), (_, v1)): (st::RowShape, st::RowShape)) -> Collection<'s, Time, CC> {
@@ -333,7 +363,16 @@ impl Backend for CorgiBackend {
         // compiles the projection once, now, at both inputs' shapes, and emits corgi columns
         // directly as `CorgiContainer`s — column-native, no row round-trip.
         let backend = CorgiJoinBackend::new(projection, [k, v0, v1]);
-        join_with_tactic::<_, _, _, CC>(l, r, "Join", ProxyJoinTactic::new(backend)).as_collection()
+        // An import is read as of its time. Reading one input as of a time `s` or both gives the
+        // same output when the other's times are at or beyond `s` (`(t0 ∨ s) ∨ t1 = t0 ∨ t1`
+        // then), as every program time is; two imports share one time. So the join reads both
+        // inputs as of the join of the two times, a one-element frontier as `Arr` says.
+        let since = differential_dataflow::lattice::Lattice::join(&l.since, &r.since);
+        if since == Time::minimum() {
+            join_with_tactic::<_, _, _, CC>(l.arranged, r.arranged, "Join", ProxyJoinTactic::new(backend)).as_collection()
+        } else {
+            join_with_tactic::<_, _, _, CC>(l.arranged, r.arranged, "Join", ProxyJoinTactic::new(BackendFrontier::new(backend, Antichain::from_elem(since)))).as_collection()
+        }
     }
 
     fn half_join<'s>(stream: Collection<'s, Time, CC>, arrangement: Self::Arr<'s>, strict: bool, projection: &Lambda, level: usize, ((k, v0), (_, v1)): (st::RowShape, st::RowShape)) -> Collection<'s, Time, CC> {
@@ -342,16 +381,31 @@ impl Backend for CorgiBackend {
         // (`CorgiHalfJoinBackend`); dogs3's proxy half-join tactic decides which pairs meet.
         // `CorgiPact` places each update on the worker holding its key, as `arrange` placed the rows.
         // The stream's value is `(val, time)`, as `lift` makes it; the projection reads the val.
+        //
+        // An imported arrangement is read as it is stored, even one read as of a later time. The
+        // stream's updates are the program's, at times at or beyond that time (its inputs are at
+        // its epochs, and its imports are read as of it), and a published trace's times are
+        // `(epoch, [])`. So an arranged row before the as-of time precedes every stream update in
+        // the total order on times, strictly, as it would advanced to it; its data time joins the
+        // stream's, which is at or beyond the as-of time, to the same time either way; and the
+        // rows met by an update sum to the trace's accumulation at the update's epoch, which
+        // compaction preserves.
         let Shape::Prod(mut v0) = v0 else { panic!("half_join: a stream value of shape {v0}, not (val, time)") };
         let backend = CorgiHalfJoinBackend::new(projection, level + 1, [k, v0.swap_remove(0), v1]);
         let tactic = ProxyHalfJoinTactic::new(backend, strict);
         let yielding = |_timer: std::time::Instant, count: usize| count > 1_000_000;
-        half_join_with_tactic(stream.inner, arrangement, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic).as_collection()
+        half_join_with_tactic(stream.inner, arrangement.arranged, CorgiPact, step_back, yielding, HalfJoinBatcher::new(strict, level + 1), tactic).as_collection()
     }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer, shape: st::RowShape) -> Self::Arr<'s> {
         // Amortize columnar corrections while reusing sweep scratch across wide presentations.
-        let tactic = ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone(), shape)).with_key_batch_size(256);
-        reduce_with_tactic::<_, CTrace, _>(a, "CorgiReduce", tactic)
+        let backend = CorgiReduceBackend::new(reducer.clone(), shape);
+        // An import is read as of its time; the output is the program's own, read as it is.
+        let Arr { arranged, since } = a;
+        Arr::built(if since == Time::minimum() {
+            reduce_with_tactic::<_, CTrace, _>(arranged, "CorgiReduce", ProxyReduceTactic::new(backend).with_key_batch_size(256))
+        } else {
+            reduce_with_tactic::<_, CTrace, _>(arranged, "CorgiReduce", ProxyReduceTactic::new(BackendFrontier::new(backend, Antichain::from_elem(since))).with_key_batch_size(256))
+        })
     }
 
     fn inspect<'s>(c: Collection<'s, Time, CC>, label: String) -> Collection<'s, Time, CC> {
@@ -418,11 +472,12 @@ impl Backend for CorgiBackend {
         stream.as_collection()
     }
     fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: Scope<'r, Time>) -> Self::Arr<'r> {
-        a.enter_region(region)
+        Arr { arranged: a.arranged.enter_region(region), since: a.since }
     }
 }
 
-/// A root import for [`render_program`]: corgi containers, or a published export.
+/// A root import for [`render_program`]: corgi containers, or a published export read as of the
+/// time the program starts.
 #[allow(clippy::large_enum_variant)] // One per import, consumed as the program renders.
 pub(crate) enum CorgiImport<'s> {
     Columns(Collection<'s, Time, CC>),
@@ -451,7 +506,9 @@ pub(crate) fn render_program<'s>(
     let logger = scope.worker().logger_for::<crate::backend::LabelEventBuilder>("ddir/labels");
     let arranged = rendered.iter().zip(&s.exports).map(|(r, export)| {
         let first = scope.worker().peek_identifier();
-        let arranged = r.arrange();
+        // A re-exported import is the producer's arrangement; its own readers import it as of
+        // the times they start.
+        let arranged = r.arrange().arranged;
         if let Some(logger) = &logger {
             let end = scope.worker().peek_identifier();
             logger.log(crate::backend::LabelEvent { first, end, label: format!("export {}", export.name) });
@@ -467,7 +524,7 @@ pub(crate) type ExportTrace = TraceAgent<CTrace>;
 
 /// Arrange a collection as a published trace.
 pub(crate) fn arrange_export<'s>(c: Collection<'s, Time, CC>) -> Arranged<'s, ExportTrace> {
-    CorgiBackend::arrange(c)
+    CorgiBackend::arrange(c).arranged
 }
 
 /// The batches an export's arrangement has produced on this worker, not yet taken.
@@ -496,7 +553,8 @@ pub(crate) fn tap_export<'s>(a: &Arranged<'s, ExportTrace>, tap: ExportTap) -> t
     })
 }
 
-/// The updates of an imported columnar export, as corgi containers.
-pub(crate) fn export_containers<'s>(a: Arranged<'s, ExportTrace>) -> Collection<'s, Time, CC> {
-    chunk_containers(a.stream, "ExportContainers")
+/// The updates of an imported columnar export, as corgi containers, read as of `since` (see
+/// [`Arr`]).
+pub(crate) fn export_containers<'s>(a: Arranged<'s, ExportTrace>, since: Time) -> Collection<'s, Time, CC> {
+    chunk_containers(a.stream, "ExportContainers", since)
 }
diff --git a/interactive/src/server/bind.rs b/interactive/src/server/bind.rs
index b90663776..2706255b1 100644
--- a/interactive/src/server/bind.rs
+++ b/interactive/src/server/bind.rs
@@ -97,11 +97,12 @@ impl Server {
         // Capture the trace's containers, never forming rows.
         let buffer: Captured = Rc::new(RefCell::new(Vec::new()));
         let buffer_in = buffer.clone();
-        let Published(mut trace) = self.published(&source).expect("checked above");
+        let mut published = self.published(&source).expect("checked above");
+        let as_of = self.import_as_of();
         let shutdown = host_dataflow(worker, |_, scope| {
             use timely::dataflow::operators::{Inspect, Probe};
-            let (arranged, shutdown) = trace.import_core(scope.clone(), "BindImport");
-            crate::backend::corgi::export_containers(arranged)
+            let (containers, shutdown) = published.import_containers(scope, "BindImport", as_of);
+            containers
                 .inner
                 .inspect_core(move |event| {
                     if let Ok((_, container)) = event {
diff --git a/interactive/src/server/install.rs b/interactive/src/server/install.rs
index 2dc2e74c3..547ce4d09 100644
--- a/interactive/src/server/install.rs
+++ b/interactive/src/server/install.rs
@@ -123,6 +123,7 @@ impl Server {
         let root = &prog.root;
         let arranged_imports = self.arranged_imports;
         let tapped = self.export_taps;
+        let as_of = self.import_as_of();
         let traces = &mut self.traces;
 
         // The id this dataflow will get; captured so `drop` can remove it.
@@ -137,7 +138,7 @@ impl Server {
                 use timely::dataflow::operators::core::Input as _;
                 use timely::dataflow::operators::Probe;
                 use differential_dataflow::AsCollection;
-                use crate::backend::corgi::{export_containers, CorgiImport};
+                use crate::backend::corgi::{export_containers, Arr, CorgiImport};
                 let mut inputs: Vec<(usize, ServerInput)> = Vec::new();
 
                 // One value per root import, rendered without rows. An input converts rows to
@@ -154,13 +155,16 @@ impl Server {
                             CorgiImport::Columns(stream.as_collection().enter(scope))
                         }
                         st::Source::Trace(t) => {
-                            // The first binding point: resolve a named trace by importing it.
+                            // The first binding point: resolve a named trace by importing it, as
+                            // of the program's start.
                             let key = canonical_source_name(t);
-                            let arranged = traces.get_mut(&key).expect("validated above").import(scope);
+                            let trace = traces.get_mut(&key).expect("validated above");
+                            debug_assert!(trace.get_logical_compaction().less_equal(&as_of), "an import as of {as_of:?}, before its trace's compaction frontier");
+                            let arranged = trace.import(scope);
                             if arranged_imports {
-                                CorgiImport::Export(arranged)
+                                CorgiImport::Export(Arr { arranged, since: as_of.clone() })
                             } else {
-                                CorgiImport::Columns(export_containers(arranged))
+                                CorgiImport::Columns(export_containers(arranged, as_of.clone()))
                             }
                         }
                         st::Source::Parent(_) => unreachable!("root import from a parent scope"),
diff --git a/interactive/src/server/mod.rs b/interactive/src/server/mod.rs
index 258f91c16..a34465c1f 100644
--- a/interactive/src/server/mod.rs
+++ b/interactive/src/server/mod.rs
@@ -98,18 +98,21 @@ use self::sources::{canonical_source_name, clock_rows, generated_shape, InputCon
 pub struct Published(pub crate::backend::corgi::ExportTrace);
 
 impl Published {
-    /// Import into `scope` as containers, one per chunk of each batch, with the import's shutdown
-    /// button. Each worker reads its own share.
+    /// Import into `scope` as containers, one per chunk of each batch, read as of `as_of` (see
+    /// [`Server::import_as_of`]), with the import's shutdown button. Each worker reads its own
+    /// share.
     pub fn import_containers<'s>(
         &mut self,
         scope: timely::dataflow::Scope<'s, HostTime>,
         name: &str,
+        as_of: HostTime,
     ) -> (
         differential_dataflow::Collection<'s, HostTime, HostContainer>,
         ShutdownButton<CapabilitySet<HostTime>>,
     ) {
+        debug_assert!(self.0.get_logical_compaction().less_equal(&as_of), "an import as of {as_of:?}, before its trace's compaction frontier");
         let (arranged, shutdown) = self.0.import_core(scope, name);
-        (crate::backend::corgi::export_containers(arranged), shutdown)
+        (crate::backend::corgi::export_containers(arranged, as_of), shutdown)
     }
 }
 
@@ -278,6 +281,17 @@ impl Server {
         self.epoch
     }
 
+    /// The time a program installed now, or a binding, snapshot or tail started now, reads the
+    /// published traces it imports as of: the epoch before the current one, as a program time.
+    ///
+    /// [`Server::tick`] compacts every published trace to it, so their accumulations are correct
+    /// from it on; before it a trace's batches can hold counts no time had (a retraction advanced
+    /// to it beside the insertion it retracts), on which an iterative scope need not converge.
+    /// Every import advances the times it reads to it, as DD's `import_frontier` reads a trace.
+    pub fn import_as_of(&self) -> HostTime {
+        host_time(self.epoch.saturating_sub(1))
+    }
+
     /// Whether `name` (canonical) is published.
     fn is_published(&self, name: &str) -> bool {
         self.traces.contains_key(name)
diff --git a/interactive/src/server/read.rs b/interactive/src/server/read.rs
index ec525af68..060c0be83 100644
--- a/interactive/src/server/read.rs
+++ b/interactive/src/server/read.rs
@@ -42,7 +42,8 @@ impl Server {
 
         let name = canonical_source_name(name);
         let epoch = self.epoch;
-        let Published(mut trace) = self.published(&name).ok_or_else(|| format!("no trace {:?}", name))?;
+        let mut published = self.published(&name).ok_or_else(|| format!("no trace {:?}", name))?;
+        let as_of = self.import_as_of();
         let shape = self.shapes[&name].clone();
         let parts: Rc<RefCell<Vec<Updates>>> = Rc::new(RefCell::new(Vec::new()));
         let parts_in = parts.clone();
@@ -50,8 +51,8 @@ impl Server {
         let id = worker.next_dataflow_index();
         host_dataflow(worker, |_, scope| {
             // The dataflow is dropped once drained, which releases the import.
-            let (arranged, _shutdown) = trace.import_core(scope.clone(), "SnapshotImport");
-            crate::backend::corgi::export_containers(arranged)
+            let (containers, _shutdown) = published.import_containers(scope, "SnapshotImport", as_of);
+            containers
                 .inner
                 .unary::<timely::container::CapacityContainerBuilder<HostContainer>, _, _, _>(crate::runtime::exchange::GatherPact, "Snapshot", move |_, _| {
                     move |input, _output| {
diff --git a/interactive/tests/difftest.rs b/interactive/tests/difftest.rs
index 5977650fe..968abce84 100644
--- a/interactive/tests/difftest.rs
+++ b/interactive/tests/difftest.rs
@@ -21,8 +21,8 @@
 //! - `bind`: the consumer's inputs bound to the producer's published traces instead.
 //! - `reinstall`: the consumer dropped, and installed again later.
 //!
-//! Until imports read their traces as of their install, a consumer is installed with its producer,
-//! and `reinstall` does not run: see [`LATE_INSTALLS`].
+//! A consumer may be installed after its producer has run some epochs, reading compacted history
+//! as of its install: see [`LATE_INSTALLS`].
 //! - `explain`: every row of the first export is regenerated by its explanation alone (see
 //!   `Case::explainable` for the cases it skips).
 //!
@@ -77,9 +77,9 @@ enum Sh { Int, Unit, Tup(Vec<Sh>), List(Box<Sh>), Sum, Float }
 
 /// Whether `split` and `bind` may install their consumer after the producer has run some epochs,
 /// and whether `reinstall` runs. A consumer installed late imports traces whose history has been
-/// compacted, and does not converge until each import reads its trace as of the time it starts
-/// (#982). Turning this on changes no case: the late epoch is drawn either way.
-const LATE_INSTALLS: bool = false;
+/// compacted, which it reads as of the time it starts (`Server::import_as_of`). Turning this off
+/// changes no case: the late epoch is drawn either way.
+const LATE_INSTALLS: bool = true;
 
 /// The one declared sum, `S`: its lanes' payload shapes are `int`, `(int, int)` and `()`.
 const SUM_DECL: &str = "type S = A int | B (int, int) | C;";
@@ -1420,6 +1420,22 @@ fn now_secs() -> u64 { std::time::SystemTime::now().duration_since(std::time::UN
 /// Note progress, for the watchdog.
 fn tick() { LAST.store(now_secs(), std::sync::atomic::Ordering::Relaxed); }
 
+/// Seeds whose consumer, installed after its producer had run some epochs, once failed to
+/// converge: it read its imports' compacted history at the stored times, where a retraction
+/// advanced to the compaction frontier sits beside the insertion it retracts. Each now reads its
+/// imports as of its install.
+#[test]
+fn late_install_regressions() {
+    let oracles = [Oracle::Split, Oracle::Bind, Oracle::Reinstall];
+    for seed in [40000075, 40000160, 40000519] {
+        match check(&generate(seed), &oracles) {
+            Outcome::Pass => {}
+            Outcome::Refused(e) => panic!("seed {seed} refused: {e}"),
+            Outcome::Fail(oracle, why) => panic!("seed {seed}, {} oracle: {why}", oracle.name()),
+        }
+    }
+}
+
 #[test]
 #[ignore = "differential sweep; run with --release -- --ignored --nocapture"]
 fn difftest() {
diff --git a/interactive/tests/server_late_import.rs b/interactive/tests/server_late_import.rs
new file mode 100644
index 000000000..0ea2b95f0
--- /dev/null
+++ b/interactive/tests/server_late_import.rs
@@ -0,0 +1,151 @@
+//! A program installed late reads the published traces it imports as of the time it starts
+//! (`Server::import_as_of`), the traces' compaction frontier: every update's time advanced to it.
+//! Before that frontier a compacted trace's batches can hold counts no time ever had (a retraction
+//! advanced past the frontier, the insertion it retracts left behind), and an iterative scope
+//! reading them there need not converge. Imports read as containers, as rows of arrangements, and
+//! as arrangements read directly by a join and a reduce, all from the frontier on. (Found by
+//! `difftest`.)
+
+use std::sync::mpsc;
+use std::time::Duration;
+
+use interactive::server::Server;
+use interactive::{lower, parse};
+
+const PRODUCER: &str = r#"
+    export "seeds" = input 0 : ((int, int), ());
+    export "edges" = input 1 : ((int, int), ());
+"#;
+
+/// A row `(a, b)` of `x` meets an equal row of `edges` and steps to `(b % 5, b)`: `(5, 5)` steps
+/// to `(0, 5)`, which steps to itself while `edges` holds `(0, 5)`.
+const CONSUMER: &str = r#"
+    let seeds = import "seeds";
+    let edges = import "edges";
+    s: {
+        let step = x | join(edges, ((_, b), (), ()) -> (((b, 5) rem, b), ()));
+        var x = (seeds + step) | distinct;
+    }
+    export "result" = s::x;
+"#;
+
+/// The feeds, per epoch, as `(input, rows)`: `(0, 5)` is added at epoch 2 and retracted at 14, and
+/// enough else changes that the edges trace merges its batches unevenly.
+const FEEDS: &[(u64, usize, &str)] = &[
+    (0, 1, "0,0\n0,3 diff=2"),
+    (1, 1, "-1,4 diff=2"),
+    (2, 1, "1,2\n0,5"),
+    (3, 1, "4,3\n3,0\n0,4"),
+    (4, 1, "4,3 diff=2\n1,5"),
+    (5, 0, "5,5"),
+    (7, 1, "1,2\n1,5 diff=-1\n4,3 diff=-1"),
+    (8, 1, "1,2 diff=-1\n4,3 diff=-1"),
+    (10, 1, "5,3\n0,3 diff=-1"),
+    (11, 1, "2,3\n4,4\n-1,-1\n1,2 diff=-1"),
+    (14, 1, "5,5\n0,5 diff=-1"),
+    (21, 0, "-1,2"),
+];
+
+/// The consumer's rows after the last epoch, installed before epoch `late`, with imports as
+/// containers or as arrangements.
+fn run(late: u64, arranged: bool) -> Vec<(String, String, i64)> {
+    timely::execute_directly(move |worker| {
+        let program = |src: &str| { let mut p = lower::lower_tree(parse::pipe::parse(src)); p.optimize(); p };
+        let mut server = Server::new();
+        server.set_arranged_imports(arranged);
+        server.install(worker, "producer", &program(PRODUCER)).unwrap();
+        for epoch in 0..22 {
+            if epoch == late { server.install(worker, "consumer", &program(CONSUMER)).unwrap(); }
+            for (_, input, rows) in FEEDS.iter().filter(|(e, _, _)| *e == epoch) {
+                server.feed("producer", *input, rows, None).unwrap();
+            }
+            server.tick(worker);
+        }
+        server.snapshot(worker, "result").unwrap().rows()
+    })
+}
+
+#[test]
+fn late_importers_converge() {
+    // Without the fix, the late cases never finish their first tick: fail rather than hang.
+    let (send, recv) = mpsc::channel();
+    std::thread::spawn(move || {
+        let early = run(0, false);
+        let late: Vec<_> = [(17, false), (21, false), (17, true), (21, true)].into_iter().map(|(at, arranged)| run(at, arranged)).collect();
+        send.send((early, late)).unwrap();
+    });
+    let (early, late) = recv.recv_timeout(Duration::from_secs(60)).expect("a late importer did not converge");
+    assert_eq!(early.len(), 3, "(5, 5), (0, 5) and (-1, 2): {early:?}");
+    for rows in late {
+        assert_eq!(rows, early);
+    }
+}
+
+/// As above, with imports read as arrangements and operators reading them directly. `r` joins the
+/// imported `pairs` with itself: read from before the compaction frontier, the join met the two
+/// histories at their stored times and gave `r` rows there, and the loop then joined it with the
+/// imported `weights` where that history holds counts no time had. `w` reduces an import directly.
+const ARRANGED_PRODUCER: &str = r#"
+    export "pairs" = input 0 : ((int, int), ());
+    export "weights" = input 5 : (int, float);
+"#;
+
+const ARRANGED_CONSUMER: &str = r#"
+    let pairs = import "pairs";
+    let weights = import "weights";
+    let r = pairs | join(pairs, ((a, b), (), _) -> ((b, (a, b) min) eq, 0)) | distinct;
+    s: {
+        let step = x | join(weights, (k, v, _) -> (k, v));
+        var x = (r + step) | min;
+    }
+    export "result" = s::x;
+    export "least" = weights | min;
+"#;
+
+const ARRANGED_FEEDS: &[(u64, usize, &str)] = &[
+    (0, 0, "1,0 diff=1"),
+    (0, 5, "-1 val=-0.0 diff=2\n5 val=-0.0 diff=1\n3 val=-1.5 diff=1\n4 val=-1.5 diff=1\n3 val=1.5 diff=2"),
+    (2, 5, "1 val=1e300 diff=1\n0 val=-1.5 diff=1\n4 val=-0.0 diff=1\n3 val=4.0 diff=1"),
+    (4, 5, "5 val=2.5 diff=1"),
+    (5, 5, "3 val=1e300 diff=1\n-1 val=-1e-300 diff=1\n3 val=0.1 diff=1"),
+    (6, 5, "1 val=4.0 diff=2\n0 val=-1e-300 diff=2\n-1 val=1.5 diff=1\n5 val=0.1 diff=1"),
+    (7, 5, "1 val=-1e-300 diff=2\n4 val=0.0 diff=1\n2 val=0.0 diff=1\n0 val=0.0 diff=1\n1 val=4.0 diff=-1"),
+    (9, 5, "4 val=2.5 diff=1\n5 val=2.0 diff=1\n4 val=0.0 diff=1"),
+    (12, 5, "1 val=1e300 diff=1\n0 val=-1e-300 diff=-1"),
+    (14, 5, "2 val=1.5 diff=1\n-1 val=1.5 diff=-1\n4 val=0.0 diff=-1"),
+    (16, 5, "4 val=0.1 diff=1\n4 val=-1e-300 diff=1"),
+    (18, 5, "2 val=0.1 diff=1\n3 val=-3.25 diff=2\n-1 val=0.0 diff=1\n3 val=0.0 diff=1\n0 val=1.5 diff=1"),
+    (20, 5, "3 val=1.5 diff=1\n3 val=0.0 diff=1\n0 val=1.5 diff=1"),
+    (22, 5, "2 val=0.1 diff=1\n2 val=-3.25 diff=2\n1 val=2.0 diff=1"),
+    (23, 5, "5 val=-0.0 diff=2\n0 val=0.1 diff=1\n4 val=0.1 diff=1\n1 val=-1e-300 diff=-1\n5 val=0.1 diff=-1"),
+    (25, 5, "1 val=2.0 diff=1"),
+];
+
+/// The arranged consumer's exports after the last epoch, installed before epoch `late`.
+fn run_arranged(late: u64) -> (Vec<(String, String, i64)>, Vec<(String, String, i64)>) {
+    timely::execute_directly(move |worker| {
+        let program = |src: &str| { let mut p = lower::lower_tree(parse::pipe::parse(src)); p.optimize(); p };
+        let mut server = Server::new();
+        server.set_arranged_imports(true);
+        server.install(worker, "producer", &program(ARRANGED_PRODUCER)).unwrap();
+        for epoch in 0..27 {
+            if epoch == late { server.install(worker, "consumer", &program(ARRANGED_CONSUMER)).unwrap(); }
+            for (_, input, rows) in ARRANGED_FEEDS.iter().filter(|(e, _, _)| *e == epoch) {
+                server.feed("producer", *input, rows, None).unwrap();
+            }
+            server.tick(worker);
+        }
+        (server.snapshot(worker, "result").unwrap().rows(), server.snapshot(worker, "least").unwrap().rows())
+    })
+}
+
+#[test]
+fn late_arranged_importers_converge() {
+    let (send, recv) = mpsc::channel();
+    std::thread::spawn(move || {
+        send.send((run_arranged(0), run_arranged(25))).unwrap();
+    });
+    let (early, late) = recv.recv_timeout(Duration::from_secs(30)).expect("the late importer did not converge");
+    assert_eq!(early.0.len(), 1, "key 1, at the least value 0: {:?}", early.0);
+    assert_eq!(late, early);
+}

From 4c5cfbe02845cf77a36eac8ca737a8c9ce5960fd Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 24 Oct 2026 00:15:03 -0400
Subject: [PATCH 2/3] CI: a short late-install difftest sweep

Programs installed late read every import as of the time they start. The
release job now runs 500 late-install seeds under the split, bind and
reinstall oracles, about five seconds once built, beside the regression
test of the three seeds that used to hang, which runs in every cargo test.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o
---
 .github/workflows/test.yml | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 29f6fc9fd..e31a24406 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -91,6 +91,12 @@ jobs:
         run: |
           python3 interactive/bench/ldbc/authored/check.py --server target/release/ddir_server \
             --fixture --workers 4 --reps 1 --warmup 0 --output /tmp/ldbc-authored-check
+      # Programs installed late read every import as of the time they start. A late install
+      # that reads an import at its stored times can hang (no progress for 120 s) or disagree.
+      - name: Late-install difftest sweep
+        run: |
+          DIFFTEST_SEED=40000000 DIFFTEST_CASES=500 DIFFTEST_ORACLES=split,bind,reinstall \
+            cargo test --release -p interactive --test difftest -- --ignored --nocapture
       # The explanation suite's heavier sweeps (soundness regression, fuzz)
       # are #[ignore]d for local debug runs but must gate merges. The metric
       # report is skipped (it prints, it does not assert).

From 4225371178eab9aa9cb58606883f843d45eec8e1 Mon Sep 17 00:00:00 2001
From: Frank McSherry <fmcsherry@me.com>
Date: Sat, 24 Oct 2026 01:24:07 -0400
Subject: [PATCH 3/3] DDIR: wrap every join and reduce backend, not only
 imports'

An arrangement the program built has `since` at the minimum, where
`BackendFrontier` returns after one comparison per unit. Wrapping every join
and reduce, rather than branching on the minimum, compiles one operator for
each instead of two.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019XsakwFGwtESwZ1gSYwG4o
---
 interactive/src/backend/corgi.rs | 17 +++++++----------
 1 file changed, 7 insertions(+), 10 deletions(-)

diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index 5a38936c5..854f2206e 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -367,12 +367,11 @@ impl Backend for CorgiBackend {
         // same output when the other's times are at or beyond `s` (`(t0 ∨ s) ∨ t1 = t0 ∨ t1`
         // then), as every program time is; two imports share one time. So the join reads both
         // inputs as of the join of the two times, a one-element frontier as `Arr` says.
+        // Arrangements the program built have `since` at the minimum, where the wrapper returns
+        // after one comparison per unit, so every join is wrapped and compiles to one operator.
         let since = differential_dataflow::lattice::Lattice::join(&l.since, &r.since);
-        if since == Time::minimum() {
-            join_with_tactic::<_, _, _, CC>(l.arranged, r.arranged, "Join", ProxyJoinTactic::new(backend)).as_collection()
-        } else {
-            join_with_tactic::<_, _, _, CC>(l.arranged, r.arranged, "Join", ProxyJoinTactic::new(BackendFrontier::new(backend, Antichain::from_elem(since)))).as_collection()
-        }
+        let backend = BackendFrontier::new(backend, Antichain::from_elem(since));
+        join_with_tactic::<_, _, _, CC>(l.arranged, r.arranged, "Join", ProxyJoinTactic::new(backend)).as_collection()
     }
 
     fn half_join<'s>(stream: Collection<'s, Time, CC>, arrangement: Self::Arr<'s>, strict: bool, projection: &Lambda, level: usize, ((k, v0), (_, v1)): (st::RowShape, st::RowShape)) -> Collection<'s, Time, CC> {
@@ -400,12 +399,10 @@ impl Backend for CorgiBackend {
         // Amortize columnar corrections while reusing sweep scratch across wide presentations.
         let backend = CorgiReduceBackend::new(reducer.clone(), shape);
         // An import is read as of its time; the output is the program's own, read as it is.
+        // As for the join, the wrapper at the minimum is one comparison per unit.
         let Arr { arranged, since } = a;
-        Arr::built(if since == Time::minimum() {
-            reduce_with_tactic::<_, CTrace, _>(arranged, "CorgiReduce", ProxyReduceTactic::new(backend).with_key_batch_size(256))
-        } else {
-            reduce_with_tactic::<_, CTrace, _>(arranged, "CorgiReduce", ProxyReduceTactic::new(BackendFrontier::new(backend, Antichain::from_elem(since))).with_key_batch_size(256))
-        })
+        let backend = BackendFrontier::new(backend, Antichain::from_elem(since));
+        Arr::built(reduce_with_tactic::<_, CTrace, _>(arranged, "CorgiReduce", ProxyReduceTactic::new(backend).with_key_batch_size(256)))
     }
 
     fn inspect<'s>(c: Collection<'s, Time, CC>, label: String) -> Collection<'s, Time, CC> {