#Differential Dataflow for MoonBit — work in progress

• Timestamp lattices, product partial orders, antichains, and counted frontiers.
• Counted cyclic progress, nested feedback loops, custom timestamp refinements and identity regions.
• Output capabilities, custom unary operators, and frontier notifications.
• Signed and presence-only differences and generic update consolidation.
• Sealed trace batches, logical/physical compaction, and snapshot cursors.
• Buffered input sessions, cooperative workers, progress probes, and fanout.
• Worker groups with keyed exchange, global progress, and nested peer iteration.
• Map, flat-map, filter, explode, negate, concat, and consolidation.
• Weighted incremental joins, semijoins, and antijoins, including retractions.
• Grouped reduction, count, threshold, and distinct with synthetic partial-order timestamps.
• Shared arrangements and independent trace readers, live imports, historical lookups, and total-order upserts with upstream tie-breaking.
• Capture/replay with duplicate suppression, out-of-order messages, counted progress, retryable writers and externally activated sources.
• Incremental BFS distances, minimum-label propagation, strongly connected component edges, bidirectional shortest-path queries, and sequential neighbor-state updates.
• Exclusive prefix sums at sparse 64-bit positions, with independent keys and dynamic queries.
• Stable identifier assignment with iterative collision repair and Rust-compatible FNV encodings.

#Installation

moon add bobzhang/differential@0.1.0

#Example: totals that change when prices change

///|
test "reprice existing orders incrementally" {
  let worker = @differential.Worker()
  let (orders, order_stream) : (
    @differential.InputSession[(Int32, String), Int64, Int64],
    @differential.Collection[(Int32, String), Int64, Int64],
  ) = worker.input(0)
  let (prices, price_stream) : (
    @differential.InputSession[(Int32, Int64), Int64, Int64],
    @differential.Collection[(Int32, Int64), Int64, Int64],
  ) = worker.input(0)
  let changes = []
  let probe = order_stream
    .join_map(price_stream, fn(_, customer, price) { (customer, price) }, multiply=fn(
      quantity,
      present,
    ) {
      quantity * present
    })
    .reduce(fn(_, values) {
      let mut total = 0L
      for (price, quantity) in values {
        total += price * quantity
      }
      [(total, 1L)]
    })
    .inspect(fn(data, time, diff) { changes.push((data, time, diff)) })
  orders.update((1, "Alice"), 3)
  prices.insert((1, 7))
  orders.advance_to(1)
  prices.advance_to(1)
  orders.flush()
  prices.flush()
  assert_true(worker.run() is Quiescent(_))
  prices.remove((1, 7))
  prices.insert((1, 9))
  orders.close()
  prices.close()
  assert_true(worker.run() is Quiescent(_))
  assert_true(probe.done())
  assert_eq(changes, [
    (("Alice", 21L), 0L, 1L),
    (("Alice", 21L), 1L, -1L),
    (("Alice", 27L), 1L, 1L),
  ])
  worker.dispose()
}

#Example: incremental fixed points

///|
test "fixed point updates after a root changes" {
  let worker = @differential.Worker()
  let scope = worker.scope(0L, @differential.SummaryDomain::int64())
  let (input, roots) : (
    @differential.InputSession[Int32, Int64, Int64],
    @differential.Collection[Int32, Int64, Int64],
  ) = scope.input()
  let reachable = scope.iterate(roots, fn(_, values) {
    values
    .concat(
      values.flat_map(fn(node) { if node < 5 { [node + 1] } else { [] } }),
    )
    .distinct()
  })
  let counts : Map[Int32, Int64] = Map([])
  let probe = reachable.inspect(fn(node, _, diff) {
    counts[node] = counts.get(node).unwrap_or(0) + diff
  })
  input.insert(2)
  input.advance_to(1)
  input.flush()
  assert_true(worker.run() is Quiescent(_))
  assert_true(!probe.less_than(1))
  assert_eq(counts.get(3), Some(1L))
  input.remove(2)
  input.insert(4)
  input.close()
  assert_true(worker.run() is Quiescent(_))
  assert_true(probe.done())
  assert_eq(counts.get(3), Some(0L))
  assert_eq(counts.get(5), Some(1L))
  worker.dispose()
}

#Current contracts

#Dataflow lifecycle

#Runnable upstream algorithms

#Persistent capture

#Custom trace storage

#Message times and record times

• inspect_batch observes the capability time and a copied record array; inspect_container also observes frontier changes and forwards the stream.
• results_in(step, domain) changes logical record times and drops overflows, while retaining the original message capability.
• delay(advance) changes both times and holds messages until the input frontier passes their delayed capability. The function must be pure, monotone, and only advance times.
• join_function joins input times with computed row times and applies an explicit weight multiplication callback.
• consolidate_stream consolidates each message immediately, without waiting for all updates at a logical time.
• OperatorInput.for_each_batch exposes original messages and borrowed input capabilities; for_each remains a convenience API grouped by record time.

#Min-plus shortest paths

#Reducers with explicit corrections

#Runtime-defined iteration depths

#Validation

moon fmt --check
moon check --target all --warn-list +unnecessary_annotation --deny-warn
moon test --target all
# Restore pinned references using UPSTREAM.md first:
python3 validation/compare.py
python3 validation/cluster_compare.py
python3 validation/transport_failures.py
python3 validation/journal_check.py
# Optional Clang ASan/LSan validation:
python3 validation/sanitize_cluster.py

• 10,240 frontier/compaction rounds and 1,024 live weighted join epochs.
• 128 product-time reduction cases, with 3,200 independent snapshot queries.
• 768 streamed reduction epochs with independently advanced antichain capabilities.
• 256 feedback epochs and 256 nested fixed-point epochs with a cyclic graph BFS oracle.
• 1,024 shared-arrangement epochs with independent reader compaction.
• 8,192 upsert records, including equal-time conflicts and deletion.
• 256 reordered/duplicated capture-replay cases, comparing exact progress transitions.
• 128 eager source cases against Rust, with records available before sealed progress.
• 1,536 activated capture/replay epochs with update retries, duplicate messages, progress arriving before data, and an independent accumulated-state oracle.
• 512 partitioned join epochs against Rust running two workers.
• 128 distributed replay epochs in two/three native processes against Rust running three workers, including duplicates across peers and delayed data.
• 128 TCP process join epochs and 128 fixed-point epochs with one/two nested loops and two/three peers.
• 192 dynamic graph epochs comparing BFS, label propagation, SCC edge weights, and scheduled variants.
• 2,048 cases for 128-bit wrapping arithmetic, checked overflow, and Duration carries.
• 192 prefix-sum epochs with sparse positions, data/query retractions and multiplicities.
• 384 identifier epochs with forced collisions, winner deletions and FNV encoding checks.
• 192 bidirectional search/sequential-update epochs with independent graph recomputation.