Greptile Summary

This PR enables append pipelining by default in s2-lite, completing the work tracked in issue #48. It decouples DB write submission (now using await_durable: false) from durability acknowledgement, introduces a shared DurabilityNotifier that fans out SlateDB's watch-channel status to per-append callbacks, and replaces the per-streamer semaphore with a backend-wide one — allowing backpressure to be budgeted across all streams via the new --append-inflight-bytes flag (default 128 MiB).

Key points from the review:

• Architecture is sound: serialised write submission + parallel durability waiting is the correct design for pipelining safety; CommandState applied-point tracking correctly gates trim/fence writes.
• DurabilityNotifier implementation is clean: callbacks are always fired outside the lock, waiter insertion maintains sorted order, and all edge cases (already-durable, already-closed, out-of-order subscriptions) are covered by tests.
• Implicit semaphore capacity invariant: the old append_permit code clamped num_permits to the semaphore's max capacity to prevent acquire_many from blocking forever. The new code drops this clamp, relying implicitly on RECORD_BATCH_MAX bounding incoming batch sizes and on the semaphore being created with at least RECORD_BATCH_MAX capacity. A debug_assert! would make this assumption explicit.
• --append-inflight-bytes lacks an env = attribute: unlike other LiteArgs fields, there is no env-var alias, which may surprise operators who previously configured behaviour via S2LITE_PIPELINE.
• Redundant log line: args.append_inflight_bytes is logged twice in server.rs; the first (message-less) info! call appears to be a leftover.
• Test coverage is thorough: unit tests for DurabilityNotifier, new Streamer unit tests for durability-gated ack ordering and durable-seq jumps, and an integration test asserting ack/tail monotonicity and read order under full pipelining.

Confidence Score: 4/5

• PR is safe to merge; the pipelining logic is correct and well-tested, with only minor style concerns remaining.
• The core durability-gating design is sound: writes are serialised, acks are only released after the DB confirms durability, and the notifier correctly fans out to all waiters. The implicit semaphore capacity invariant (removed upper-bound clamp) is not a practical risk given server-side batch size validation, but it is undocumented. The missing env-var alias and duplicate log line are cosmetic. No correctness or security issues found.
• lite/src/backend/streamer.rs (implicit permit-cap invariant) and lite/src/server.rs (missing env var, duplicate log)

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant Client
    participant StreamerClient
    participant Streamer
    participant DB as SlateDB
    participant DurabilityNotifier

    Client->>StreamerClient: append_permit(input)
    StreamerClient->>StreamerClient: acquire_many(metered_size) on shared semaphore
    StreamerClient-->>Client: AppendPermit

    Client->>StreamerClient: permit.submit()
    StreamerClient->>Streamer: Message::Append (via msg_tx)
    Streamer->>Streamer: handle_append → sequence_records
    Streamer->>Streamer: push db_submit_append future → db_writes_pending

    Streamer->>DB: write_with_options(wb, await_durable=false)
    DB-->>Streamer: WriteHandle { seqnum: db_seq }
    Streamer->>Streamer: push InFlightAppend { db_seq, records } → inflight_appends
    Streamer->>DurabilityNotifier: subscribe(db_seq, callback)

    Note over Streamer: Multiple in-flight appends allowed here

    DB->>DurabilityNotifier: db.subscribe() watch fires (durable_seq advances)
    DurabilityNotifier->>DurabilityNotifier: notify_waiters → fire callbacks
    DurabilityNotifier->>Streamer: Message::DurabilityStatus(Ok(durable_seq)) (via msg_tx)

    Streamer->>Streamer: on_db_durable_seq_advanced(durable_seq)
    Streamer->>Streamer: pop inflight_appends → stable_pos, follow_tx.send, pending_appends.on_stable

    Streamer-->>Client: AppendAck (via oneshot reply_tx)
    StreamerClient->>StreamerClient: drop(sema_permit) — releases semaphore

_{Last reviewed commit: 7a8e490}