Streaming Uploads, SSE Resumption, and Disconnect Policy

Long AI work has two different streams:

• large input bytes moving toward storage and validation;
• small ordered progress events moving toward a client.

They need different protocols and failure policies. Upload bytes should not sit in RAM or in an in-memory job message. Progress should not be a lossy broadcast pretending to be history. A client disconnect should not accidentally decide whether expensive durable work lives or dies.

The companion extends mosaic-api with a resumable SSE event stream backed by a bounded in-memory journal plus live broadcast. It is intentionally not called durable: the following SQLx chapters move the same cursor contract onto transactional storage.

Learning objectives

• stream request bodies without collecting unbounded bytes;
• choose proxy upload versus direct object-store upload;
• design an upload state machine with hashes and finalization;
• understand SSE framing, IDs, keep-alives, and reconnection;
• combine replay history with a live event channel without gaps or duplicates;
• detect expired cursors and lag explicitly;
• separate client connection lifetime from run lifetime;
• bound slow consumers by events and bytes;
• test replay, live lag recovery, cursor expiry, and disconnect behavior.

Dependencies and mental model

Depends on Axum/Tower, Tokio channels, domain identities, cancellation, and backpressure.

media plane
client ── bytes ──► object storage/staging
                       │ verify size/hash/type
                       ▼
                  immutable artifact
                       │ artifact ID
                       ▼
control plane       durable run/job
                       │ ordered events
                       ▼
client ◄─ SSE replay + live tail

The control plane carries identifiers, policy, and small metadata. The data plane carries bounded media. Keeping them separate reduces memory pressure, retry ambiguity, and authority.

Never collect an unbounded body

Axum’s Body::into_data_stream yields byte chunks. axum::body::to_bytes(body, limit) is safe for small known payloads only when the limit is deliberate. For media, stream:

let mut stream = body.into_data_stream();
while let Some(chunk) = stream.next().await {
    let chunk = chunk?;
    received = received
        .checked_add(chunk.len())
        .ok_or(UploadError::TooLarge)?;
    if received > maximum_bytes {
        return Err(UploadError::TooLarge);
    }
    hasher.update(&chunk);
    sink.write_all(&chunk).await?;
}

This bounds buffering, not total disk/object usage. The sink, hash, multipart state, and downstream decoder need their own limits.

Do not trust Content-Length as the only check:

• it may be absent for chunked transfer;
• a client may lie;
• compression changes representation size;
• multipart has per-part and aggregate overhead;
• image/video decode can expand far beyond wire bytes.

Use the declared size for early rejection and admission reservation, then count actual streamed bytes.

Prefer direct upload for large artifacts

For large audio/video/images:

1. authenticated client requests an upload session;
2. API checks tenant quota, allowed type, maximum size, and purpose;
3. API creates an upload ID and short-lived scoped object-store authorization;
4. client uploads single or multipart bytes directly;
5. client calls finalize with expected size/hash/part evidence;
6. worker verifies object metadata and, where required, streams the object to compute the server hash;
7. object becomes immutable ready, then a run may reference its artifact ID.

initiated → uploading → uploaded_unverified → verifying → ready
                     └──────────────► failed/quarantined

Never let the caller choose an arbitrary bucket/key or finalize another tenant’s upload. The upload record binds tenant, generated key, purpose, maximum bytes, expected content type, expiration, and state.

Multipart safety

• cap part count and size;
• record provider upload ID server-side;
• require a canonical ordered part list;
• abort expired multipart uploads;
• do not expose permanent credentials;
• prevent overwriting a finalized object;
• verify total size and checksum independently;
• quarantine until media inspection completes.

Presigned authorization is bearer authority. Keep it short-lived, narrow to one key/operation, use TLS, and do not log it.

Media validation is not the Content-Type header

A safe pipeline checks:

1. wire byte count;
2. content signature/magic;
3. parser/decoder with resource limits;
4. image dimensions/pixels/frames;
5. audio channels/sample rate/duration;
6. video streams/codecs/duration/frame dimensions;
7. archive/container nesting and expansion ratio;
8. malware/policy scan where applicable;
9. canonical transcoding or metadata extraction in a sandbox;
10. immutable hash and provenance.

File extensions and MIME headers are hints. Decoders process hostile input; run native tools with least privilege, time/memory/CPU limits, no ambient network, controlled paths, and versioned toolchains.

SSE as an ordered event protocol

Server-Sent Events use an HTTP response with text/event-stream. A logical event may include:

id: 42
event: run_progress
data: {"sequence":42,"completed":7,"total":10}

The blank line terminates the event. id updates the client’s last-event ID. On reconnect, EventSource can send Last-Event-ID; non-browser clients can set it explicitly.

SSE is a good fit when:

• the server sends one-way progress;
• standard HTTP auth/proxies are desired;
• events are text/JSON and moderate rate;
• reconnect/resume matters.

Use WebSockets when you truly need a bidirectional session/protocol. Use object storage/download endpoints for large binary output. Do not base64 video frames into SSE.

Event contract

The companion event is:

pub struct RunEvent {
    pub sequence: u64,
    pub run_id: RunId,
    pub kind: RunEventKind,
    pub data: serde_json::Value,
}

sequence is monotonic within a run. Production records also need:

• globally unique event ID if cross-run addressing is needed;
• event schema version;
• persisted timestamp from a clear clock policy;
• producer/attempt identity;
• trace/program/model/artifact revisions;
• classification/redaction metadata;
• idempotent append key.

The SSE id is the sequence. The JSON repeats it so stored/replayed bodies remain self-describing.

Event kinds should represent stable facts (run_started, artifact_ready, run_completed), not ephemeral UI prose. Progress may be coalescible; terminal, billing, policy, and audit events are not.

Snapshot first, live tail second—without a race

A naïve stream:

1. query history;
2. subscribe to broadcast;

can miss an event appended between the two. Reversing blindly can duplicate an event in both replay and live.

The companion:

1. subscribes to live broadcast;
2. snapshots journal entries with sequence > cursor;
3. replays snapshot;
4. reads live;
5. discards every event whose sequence is not greater than its local cursor.

let receiver = self.live.subscribe();
let pending = self.snapshot_after(&run_id, after).await?;

EventSubscription {
    receiver,
    pending: pending.into(),
    cursor: after,
    // ...
}

This prevents gaps and deduplicates the overlap. In a multi-process durable system, subscribe/notify does not replace querying rows. Treat notification as a wake-up hint:

query events > cursor → send in order → wait for notification/poll → query again

The database sequence is truth.

Live broadcast is not history

Tokio broadcast retains a bounded ring. A slow receiver gets Lagged(n) after messages are overwritten. Sending successfully means receivers existed, not that any receiver observed the message.

The companion catches lag and refills from the journal after its cursor:

Err(RecvError::Lagged(_)) => {
    self.pending = self
        .hub
        .snapshot_after(&self.run_id, self.cursor)
        .await?
        .into();
}

This is the correct relationship:

journal/storage = replayable truth
broadcast/notify = low-latency wake-up

Never use broadcast alone for billing, terminal state, audit, or resumable progress.

Cursor retention and 410 Gone

History cannot remain forever at infinite cardinality. Define retention by event class and run lifecycle. If the client asks after sequence 0, but the oldest retained event is 19, silently starting at 19 creates a false complete history.

The companion returns:

{
  "error": {
    "code": "event_history_expired",
    "message": "resume from sequence 19 or fetch the run snapshot"
  }
}

with 410 Gone before opening SSE. If a connected client lags beyond retained history, the stream emits reset_required and closes. The client fetches a canonical run snapshot, then reconnects from the snapshot’s event cursor.

Retention policy should guarantee terminal events/snapshot availability longer than ordinary progress. A snapshot is not fabricated replay; it is a versioned current state plus cursor.

The Axum SSE handler

The companion route:

GET /v1/runs/{run_id}/events
Last-Event-ID: 17

parses the cursor, builds EventSubscription, converts events to Axum Event, and adds keep-alive comments:

Ok(Sse::new(event_stream).keep_alive(
    KeepAlive::new()
        .interval(Duration::from_secs(15))
        .text("keep-alive"),
))

Keep-alives help intermediaries keep idle connections open; they are not persisted progress and must not advance the cursor. Choose the interval below known idle timeouts with margin.

The handler returns quickly with a response stream. The ordinary two-second admission timeout does not impose a two-second lifetime on the response body. Streaming needs separate idle/write/run deadline policies.

Disconnect semantics

Ask one question per operation: who owns the work after admission?

Connection-owned work

Examples: cheap search preview, live microphone session with no durable result.

On disconnect:

• cancel children;
• stop provider/tool/media work where safe;
• join/clean up with deadline;
• record cancelled/abandoned outcome.

Durable run-owned work

Examples: accepted video analysis, report generation, training job.

On disconnect:

• drop the subscriber and its buffers;
• run continues under run deadline/policy;
• client reconnects from cursor;
• explicit authenticated cancel command changes run state.

The companion proves the second policy: dropping EventSubscription does not cancel the run, and a later completed event can still append.

Do not infer cancellation from a missing TCP client after returning 202. Do not keep unlimited per-client send buffers so a disconnected/slow consumer harms the run.

Slow-consumer policy

Bound:

• per-connection queued event count;
• serialized byte count;
• maximum event size;
• write/idle duration;
• total connections per tenant/run;
• reconnect rate.

Possible policy:

• coalesce/drop superseded progress events;
• never drop terminal/policy/audit events from storage;
• close live connection when it cannot keep up;
• client resumes from last delivered ID;
• if retention expired, snapshot/reset.

The framework/OS socket buffer is also a queue. Measure connection count, queued bytes, send latency, lag/reconnect/reset counts—not individual run IDs as metric labels.

Runnable companion implementation

cd rust/rust-ai-engineering
cargo run -q -p mosaic-api --example http_boundary
cargo test -p mosaic-api

Expected final example lines:

202 Accepted {"run_id":"run-0000000000000001","status":"queued"}
queued run-0000000000000001 for demo
replayed event 1 Queued

The thirteen API tests include:

• SSE content type and serialized id: 1;
• replay strictly after a cursor;
• live broadcast lag repaired from history;
• explicit expired cursor;
• invalid Last-Event-ID;
• subscriber drop independent from run completion;
• admission and middleware failures from the previous chapters.

Step-by-step production migration

The companion journal is process memory and bounded to 256 events per run. To make the contract durable:

1. define run_events(run_id, sequence, kind, body, created_at) with unique (run_id, sequence);
2. append event and update run projection in one transaction;
3. allocate sequence under row lock/advisory strategy or use a per-run counter;
4. commit before notifying;
5. SSE queries rows > cursor ORDER BY sequence;
6. notification wakes the loop; query closes races;
7. terminal event is immutable/idempotent;
8. compaction writes a versioned snapshot/cursor before deleting eligible progress;
9. authorization scopes every query by tenant/run;
10. reconnect/load/replica/crash tests verify no gaps.

The next storage chapters implement these transaction and lease foundations.

Failure investigation: progress jumped from 14 to 22

1. inspect SSE IDs actually received and last acknowledged cursor;
2. check client reconnect sent Last-Event-ID;
3. look for broadcast Lagged and buffer overflow;
4. query durable events 15..22;
5. verify replay/live subscription race and dedup logic;
6. determine whether retention expired and reset was handled;
7. check event producer skipped/duplicated sequences transactionally.

Do not hide the gap by relabeling 22 as 15.

Failure investigation: duplicate events after reconnect

1. compare stable run/sequence, not delivery attempt;
2. confirm query uses sequence > last_event_id, not >=;
3. verify cursor advances only after complete event delivery;
4. inspect snapshot/subscribe overlap;
5. make client reducer idempotent by (run_id, sequence);
6. test reconnect at every event boundary.

At-least-once delivery plus idempotent consumption is often simpler than claiming exactly once.

Failure investigation: upload succeeded but run cannot read media

1. distinguish provider multipart completion from application finalization;
2. verify bucket/key/version belongs to upload record and tenant;
3. compare declared, provider, and server-computed size/hash;
4. inspect object-store consistency/version selection and encryption authority;
5. confirm validation/quarantine state;
6. ensure job references immutable artifact/version, not mutable path;
7. retry verification idempotently or mark a typed failed state.

Alternatives and trade-offs

API-proxied upload

Simpler client/auth and centralized validation. Consumes API bandwidth/connections and requires careful streaming/temp cleanup. Suitable for small artifacts.

Direct object-store upload

Scales bytes separately and supports multipart/resume. Adds session/finalization/authorization and orphan cleanup complexity.

SSE

Simple one-way HTTP resume model. Text-only and browser header/auth limitations may matter.

WebSocket

Bidirectional and flexible. Requires a protocol, reconnection/resume design, backpressure, and often more proxy/operational work. It does not make events durable.

Polling

Robust through intermediaries and easy to cache. Higher latency/request load. A status snapshot plus cursor is still valuable as SSE recovery.

Security and performance implications

Security:

• scope upload authorization to tenant/key/size/type/expiry;
• validate actual bytes and decoded resource cost;
• sandbox hostile native decoders;
• authorize SSE on every connect and prevent cross-tenant run lookup;
• bound connections/reconnects/event size;
• redact event data and never stream secrets/provider internals;
• explicit cancel endpoint requires current authorization and state transition checks.

Performance:

• direct uploads keep media off API memory/bandwidth;
• hashing is sequential CPU work and can be fused with streaming;
• multipart size trades retry granularity for request overhead;
• one broadcast channel carrying all runs causes unrelated filtering; shard/topic strategy is needed at scale;
• querying per event causes database load; batch by cursor;
• coalescing progress reduces write and delivery amplification;
• terminal events and snapshot retrieval need low-latency indexes.

Exercises

1. Add an HTTP test that reconnects with Last-Event-ID: 1 and proves the next event is 2.
2. Change history capacity to two, append three events, and map an HTTP resume from 0 to 410.
3. Add a versioned RunSnapshot { state, last_sequence } and a reset client algorithm.
4. Design direct multipart upload tables and idempotent finalize transitions.
5. Implement a streaming SHA-256 file sink with byte limit and atomic rename.
6. Classify each event kind as durable, coalescible, sensitive, and retention class.
7. Load-test one slow and one fast subscriber; prove the slow one cannot delay the fast one or run.

Solution guidance

Subscribe before snapshot and discard sequence duplicates. The HTTP reconnect test should append a second event through api.events, set the standard header, and read only the first body chunk under a short test timeout. Finalization uses compare-and-set state plus object metadata/hash; repeated identical finalize returns the existing artifact, while mismatched evidence conflicts. A local file sink writes to a generated file in a confined staging directory, hashes while writing, sync_alls where durability requires it, then atomically renames only after validation.

Check your understanding

• Why is Content-Length insufficient?
• What race exists between history query and live subscribe?
• Why does a broadcast send not mean delivery?
• What does an SSE ID identify?
• When should disconnect cancel work?
• How does a lagged client recover?
• Why is 410 safer than silently starting at retained history?
• Which upload state makes an artifact eligible for a run?

Completion checklist

• Upload bytes and control metadata use appropriate planes.
• Actual streamed and decoded resource sizes are bounded.
• Upload finalization verifies immutable identity/hash.
• Every durable event has a stable run-local sequence.
• Replay and live tail cannot gap and deduplicate overlap.
• Broadcast/notification is only a wake-up mechanism.
• Expired history has a snapshot/reset contract.
• Slow consumers have count, byte, and time bounds.
• Disconnect/cancel behavior is explicit per operation.
• Router and event tests cover replay, lag, expiry, and drop.
• I ran the companion example and tests.

Authoritative references

• Axum SSE response
• Axum BodyDataStream
• Axum bounded to_bytes
• Tokio broadcast channel and lag
• HTML server-sent events standard
• Amazon S3 multipart upload overview
• OWASP File Upload Cheat Sheet