Live stream operations

Operating model

One owner advances each pipeline; every published view names its horizon.

Treat a pipeline instance as an in-memory projection of an authoritative ordered input. The projection records windows; a snapshot or comparison answers a question about that recorded state. It is not the input log.

1. Serialize ingestionChoose one deterministic event order and one writer for each pipeline instance.

2. Preserve lane identityPass the same source and partition identity on every event belonging to a lane.

3. Stabilize at recordingUse consecutive entry and exit confirmation when signal noise should not become window churn.

4. Evaluate explicitlyChoose processing position or event time, then take a snapshot or run a live comparison at a compatible horizon.

5. Publish finalityKeep provisional rows visibly separate and derive a changelog before replacing a consumer view.

6. Recover from source truthPersist inputs or materialized records outside Spanfold, and rebuild when corrections change earlier evidence.

Ordered ingestion

Processing position is commit order; source and partition define independent lanes.

Each committed event advances the pipeline position once. Window state and stabilization counts are isolated by window name, key, source, and partition. Do not merge partitions by omitting an identity that matters to the question.

C# · serialize lane-aware ingestion

var pipeline = EventPipeline
    .For<DeviceSignal>()
    .RecordWindows()
    .WithEventTime(signal => signal.OccurredAt, "source-utc")
    .TrackWindow(
        "DeviceOffline",
        signal => signal.DeviceId,
        signal => !signal.IsOnline,
        options => options.Stabilize(
            exitWhen: signal => signal.IsOnline,
            enterAfter: 2,
            exitAfter: 3));

foreach (var signal in orderedProviderEvents)
{
    pipeline.Ingest(signal, source: "provider-a", partition: "eu-west");
}

Rust · serialize lane-aware ingestion

let mut pipeline = spanfold::for_events::<DeviceSignal>()
    .record_windows()
    .with_event_time(|signal| signal.occurred_at_ticks)
    .window(
        "DeviceOffline",
        |signal| signal.device_id.clone(),
        |signal| !signal.is_online,
    )
    .stabilize(|signal| signal.is_online, 2, 3)
    .build()?;

for signal in ordered_provider_events {
    pipeline.ingest(signal, Some("provider-a"), Some("eu-west"))?;
}

Single writer

The .NET pipeline explicitly fails concurrent or re-entrant ingestion. Rust requires mutable access for ingestion. In both cases, serialize writes and avoid querying a pipeline while another owner is changing it.

Batch differences

Rust ingest_many(...) applies one source and partition to the batch. C# IngestMany(events) has no lane parameters, so use individual Ingest(...) calls when lane identity matters.

Transition callbacks

Callbacks are notifications, not durable delivery. In .NET, callback failure is reported after the event and history commit; blindly retrying that event can advance the projection twice.

Stabilization

Confirm noisy transitions before they become evidence.

Both runtimes count consecutive entry and exit observations separately for each lane. A non-matching observation resets the applicable count. The event that reaches the threshold supplies the boundary; a pending exit leaves the committed open window and roll-up membership unchanged.

Stabilization is observation-count hysteresis, not elapsed-time debouncing. It does not wait on a timer, reorder events, or decide whether late event time is acceptable. Omit it when every predicate transition should open or close immediately.

Time and horizons

Processing position and event time answer different questions.

Coordinate	Meaning	Operational use
Processing position	The monotonic order in which one pipeline committed events.	Replay order, known-at filtering, and live views tied to ingestion progress.
Event time	A timestamp or tick value selected from the event. Values intended to compare need the same clock identity; different clock contracts remain distinct.	Compare when source events claim a condition happened, even if ingestion order differs.
Live horizon	The explicit point where open evidence is clipped for one evaluation.	Make a current answer reproducible and keep horizon-dependent evidence provisional.

C# · snapshot and compare at one horizon

var horizon = TemporalPoint.ForTimestamp(now, "source-utc");
var snapshot = pipeline.History.SnapshotAt(horizon);

var live = pipeline.History
    .Compare("Live provider QA")
    .Target("provider-a", side => side.Source("provider-a"))
    .Against("provider-b", side => side.Source("provider-b"))
    .Within(scope => scope.Window("DeviceOffline", TemporalAxis.Timestamp))
    .Normalize(policy => policy.OnEventTime())
    .Using(rows => rows.Overlap().Residual().Missing())
    .RunLive(horizon);

Rust · snapshot and compare at one horizon

let horizon = TemporalPoint::timestamp_ticks(now_ticks);
let snapshot = pipeline.history().snapshot_at(horizon.clone())?;

let live = pipeline
    .history()
    .compare("Live provider QA")
    .target_source("provider-a")
    .against_source("provider-b")
    .scope(ComparisonScope::window("DeviceOffline").on_event_time())
    .normalization(ComparisonNormalizationPolicy::event_time())
    .overlap()
    .residual()
    .missing()
    .run_live(horizon);

A history snapshot is read-only and does not close or rewrite the source history. A live comparison clips open windows for the result only. Its horizon must use the same axis—and, for timestamps, the same clock contract—as the windows being evaluated.

Finality and revisions

Publish the row metadata with the row, then describe replacements as a changelog.

Final rows depend only on closed evidence. Provisional rows depend on at least one open window clipped to the live horizon. A later run can add, revise or update, and retract row-finality metadata as the source history changes.

C# · advance a consumer view

var changes = ComparisonChangelog.Create(
    previous.RowFinalities,
    current.RowFinalities);

var replayed = ComparisonChangelog.Replay(
    previous.RowFinalities,
    changes);

Rust · advance a consumer view

let changes = create_changelog(
    &previous.row_finalities,
    &current.row_finalities,
);

let replayed = replay_changelog(
    &previous.row_finalities,
    &changes,
);

Version the whole view

Store the horizon and plan identity beside the row set. A changelog compares two results; it is not a global event log and does not prove source completeness.

Apply atomically

Build the next result and changelog before replacing the published view. Consumers can then apply additions, revisions or updates, and retractions as one versioned transition.

Keep provisional visible

Do not flatten row finality into a run-level flag. One live result can contain stable closed-history rows and provisional horizon-dependent rows together.

Liveness

Turn silence into ordinary window evidence at caller-chosen horizons.

Both runtimes provide a deterministic LaneLivenessTracker. Call Observe/observe when a known lane reports, and Check/check when your scheduler reaches a horizon. Feed only the emitted state changes into a normal pipeline.

C# · record lane silence

var liveness = LaneLivenessTracker.ForLanes(
    startedAt,
    TimeSpan.FromSeconds(30),
    "provider-a",
    "provider-b");

var silencePipeline = EventPipeline
    .For<LaneLivenessSignal>()
    .RecordWindows()
    .WithEventTime(signal => signal.OccurredAt)
    .TrackWindow("LaneSilent", signal => signal.Lane, signal => signal.IsSilent);

foreach (var signal in liveness.Check(horizon))
    silencePipeline.Ingest(signal, source: "liveness");

Rust · record lane silence

let mut liveness = LaneLivenessTracker::for_lanes(
    started_at.clone(),
    30 * ticks_per_second,
    ["provider-a", "provider-b"],
)?;

let mut silence_pipeline = for_events::<LaneLivenessSignal>()
    .record_windows()
    .with_event_time(|signal| signal.occurred_at.magnitude())
    .track_window(
        "LaneSilent",
        |signal| signal.lane.clone(),
        |signal| signal.is_silent,
    )
    .build()?;

for signal in liveness.check(horizon)? {
    silence_pipeline.ingest(signal, Some("liveness"), None)?;
}

The liveness trackers own no timers, IO, distributed heartbeat discovery, or persistence. Observations and check horizons must be monotonic. Rust thresholds are magnitudes on the chosen temporal axis, so the caller owns the tick unit for timestamp tracking.

Late and out-of-order corrections

Admission and replay are separate decisions.

Event time does not change ingestion order. Neither pipeline rewinds an earlier transition when a late event arrives. Decide upstream whether a revision is admissible, update the authoritative source, then rebuild the affected projection or materialized history and run a new snapshot.

Application-owned replay

Keep stable event and revision IDs in the source log. When a correction replaces prior input, replay the affected deterministic order into a fresh pipeline rather than treating a changelog as an ingestion undo log.

Materialized-history path

Both runtimes can construct queryable history from closed and open records using WindowHistory.FromRecords(...) or WindowHistory::from_records(...). The caller remains responsible for producing coherent corrected records.

Result-level explanation

After rerunning, compare the old and new row-finality collections. The comparison changelog explains the visible row transition; it does not apply the source correction.

Bounded watermarks

.NET has an opt-in late-event decision helper; Rust does not.

BoundedWatermarkTracker is a .NET-only, in-memory, single-writer primitive that sits before a pipeline. It accepts explicit per-lane event-time progress and returns decisions; it never rewires or mutates pipeline history.

.NET source API

var watermarks = new BoundedWatermarkTracker(
    allowedLateness: TimeSpan.FromMinutes(2));

var decision = watermarks.Observe(
    laneId: "provider-a/eu-west",
    eventId: envelope.EventId,
    revisionId: envelope.RevisionId,
    eventTime: envelope.OccurredAt);

HandleWatermarkDecision(decision);

var advance = watermarks.AdvanceLane(
    "provider-a/eu-west",
    sourceProgress);

foreach (var released in advance.Released)
{
    HandleWatermarkDecision(released);
}

Rust boundary

Rust has no equivalent bounded-watermark or late-revision admission helper. Own buffering, lateness, rejection, correction retention, and lane progress upstream; feed only the chosen corrected input into a fresh projection.

The .NET helper implements its documented bounded policy: observations before a lane has reported progress are buffered, as are observations ahead of current progress. Progress derives and advances a lane watermark. Operators must handle the immediate decision and route every decision in advance.Released; decisions can be buffered, accepted, rejected, or corrected. The application-owned handler decides how accepted input and corrections reach a rebuilt projection—the tracker does not mutate a pipeline. It does not infer completeness, advance idle lanes, persist state, schedule progress, or coordinate distributed partitions. Do not copy these semantics into Rust unless your application deliberately owns that policy.

Retention and persistence

Persist what you need before trimming or replacing an in-memory projection.

Responsibility	.NET	Rust
Persist source events, checkpoints, and replay order	Caller	Caller
Persist window records, snapshots, results, and changelogs	Caller	Caller
Import materialized open and closed records	`WindowHistory.FromRecords(...)`	`WindowHistory::from_records(...)`
Remove closed history by processing-position boundary	`TrimClosedBefore(...)`; open windows and annotations remain	No equivalent history-drain API
Release inactive roll-up runtime state	`TrimInactiveState()`; recorded history remains	No equivalent public pipeline trim API

A retention boundary changes which historical questions the in-memory projection can answer. Persist or export required evidence first, keep the replay checkpoint aligned with the retained range, and never assume a result changelog can reconstruct discarded source history.

Safe consumer workflow

Hand consumers one coherent version at a time.

Capture

Choose a compatible horizon after the serialized ingestion owner reaches a known checkpoint. Record the source checkpoint, plan identity, axis, clock, and horizon.

Compute

Run the live comparison from a stable history view. Keep finality metadata and evidence IDs with every row; do not publish a provisional row as historical fact.

Transition

Build the changelog from the prior and current results, persist the new version, then atomically expose it. Make retractions first-class for caches and dashboards.

Correct

If upstream revises earlier input, rebuild from the authoritative replay boundary. Publish the rerun as another version and explain result changes with the changelog.

Retain

Trim only after the durable source, materialized evidence, and replay checkpoint cover the questions consumers are still allowed to ask.

Recover

After restart, restore from your durable records or replay the source. Spanfold does not persist a live pipeline, liveness tracker, or watermark tracker for you.

Runtime boundaries

Portable model, deliberately different operational helpers.

Both runtimes

Ordered synchronous ingestion, lane-aware windows, consecutive stabilization, event-time recording, processing-position history, live snapshots, provisional finality, comparison changelogs, materialized history adoption, and explicit liveness tracking. The adoption guide explains the current language-specific import boundaries.

.NET only

BoundedWatermarkTracker, WindowHistory.TrimClosedBefore(...), and EventPipeline.TrimInactiveState(). These are current repository source APIs and may be ahead of the published NuGet package.

Caller in both runtimes

Queue consumption, delivery guarantees, input deduplication, lane discovery, timers, checkpointing, durable persistence, distributed completeness, correction policy, replay scheduling, and atomic publication.

Live finality concepts C# ↔ Rust portability map

Keep changing evidence useful without calling it settled.