Skip to content

feat(core): streams with consumers and the hopperui module (M8) - #10

Merged
giraffesyo merged 2 commits into
canaryfrom
feat/streams
Sep 27, 2026
Merged

giraffesyo merged 2 commits into
canaryfrom
feat/streams

Conversation

@giraffesyo

@giraffesyo giraffesyo commented Sep 26, 2026 •

Copy link
Copy Markdown
Member

Streams, plus the hopperui module: #11 was merged into this branch (its base), so this PR carries both and completes M8.

hopperui (from #11)

hopperui.New(client, cfg) http.Handler (separate module): overview and queues, jobs and history with filters and paging, job detail, workflows as an SVG DAG, subscriptions, stream consumers with seek forms, clients. Actions (retry, cancel, pause, resume, seek) only with Config.Authorize; read-only otherwise. hopperui/cmd/hopperui is a standalone server. JobFilter.After for paging. Makefile, CI, dependabot and docs cover the module.

Streams

What changed

  • Log: hopper_stream_events, append-only and partitioned by day (dropped after Config.StreamRetention, 7 days by default, by the same leader maintenance as history). client.Streams().Append/AppendTx write one event: topic, optional key, payload, headers, a database-generated message ID. Events are identified by (xid, seq).
  • Consumers: hopper.Consume(workers, hopper.Consumer{Name, Pattern, Queue, Start, MaxAttempts, Metadata}, fn). A consumer is a named position on the log plus a topic pattern; the leader turns the events after it into delivery jobs of kind stream:<name> (the event's key becomes the ordering key), so a consumer gets typed handlers, retries, dead-lettering and competing workers for free. Message gains Stream and Position. New consumers start at StreamStartLatest (default) or StreamStartEarliest; Streams().Seek moves one to a position, a time, the start or now; Streams().Read pages the log; Streams().Consumers lists them.
  • No skipped late commits, no stalls: a consumer's row records the pg_snapshot it has read to. A pump takes a fresh snapshot and delivers the events of transactions visible in the new snapshot but not the old (pg_visible_in_snapshot, with an xid range to keep the index scan bounded), in (xid, seq) order; a pump that stops part way keeps both snapshots and the cursor and continues the same delta. A transaction that commits late is delivered when it commits, whatever its xid, and a long-running transaction holds nothing back. The plan's original "readers see only events below xmin" rule was tried first and dropped: any open writing transaction in the database stalled every consumer (the parallel test suite alone made appends invisible), and it would have made a long batch transaction in production delay all deliveries.
  • Wake-up: pool appends notify hopper_stream through the coalescing notifier and poke the local leader; transactional appends notify from the transaction; the leader's stream loop also pumps every PollInterval. Two leaders pumping one consumer serialize on its row (lock first, snapshot second).
  • CLI: hopper streams consumers, hopper streams seek <consumer> -earliest|-latest|-time|-position, and hopper subscriptions list (promised in the plan for M6, never added).
  • Tests: drivertest Streams on both drivers: ordering, reads by pattern and position, earliest/latest consumers, delivery metadata and ordering keys, partial pumps continuing a delta, seeks of every kind, a late commit delivered after younger transactions without being skipped or delivered twice, retention partitions. A client-level end-to-end test; CLI cases.
  • Docs: plan §4.5, §6, §7.10, §10, §13, §16; changelog; getting-started and operations sections.

hopperbench

The insert, claim and finalize paths are untouched by this PR (streams are new tables and a leader duty), so no benchmark run.

@giraffesyo giraffesyo changed the title feat(core): streams with consumers (M8) feat(core): streams with consumers and the hopperui module (M8) Sep 26, 2026
Base automatically changed from feat/workflows to canary September 26, 2026 14:54
@giraffesyo
giraffesyo enabled auto-merge (squash) September 26, 2026 14:55
@giraffesyo
giraffesyo merged commit 842d522 into canary Sep 27, 2026
8 checks passed
@giraffesyo
giraffesyo deleted the feat/streams branch September 27, 2026 02:23
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant