Worker event stream #1665

mitchhs12 · 2026-01-30T17:48:59Z

Summary

Adds opt-in Kafka event streaming for worker sync progress, enabling real-time monitoring of dataset sync jobs.

Emit lifecycle events (sync.started, sync.progress, sync.completed, sync.failed) per-table
Protobuf encoding via prost; Kafka delivery via rskafka (no librdkafka dependency)
1% throttling for progress events to emit every 1% increment; avoids noise on high-throughput chains
Configurable partition count (default: 16)
Best-effort delivery with retry backoff (1s, 5s, 60s) - Kafka failures don't fail jobs

See app-ampd-worker-events.md for full documentation.

Events are per-table (correct for Kafka partitioning), but raw_dataset couples the lifecycle timing of all tables because they share a single block stream.
Retry policy uses fixed delays (1s, 5s, 60s), max 66s wait. Silent failure after exhaustion is intentional: events are best-effort.
Per-table partitioning enables consumer parallelism; aggregation by job_id provides job-level view.
Sync ProgressCallback trait avoids async overhead; WorkerProgressCallback bridges to async via tokio::spawn.

Test plan

Unit tests for retry backoff and error handling
Integration tests for all event types
E2E test with Anvil verifying full lifecycle (3 tables × lifecycle events)

mitchhs12 · 2026-01-30T18:26:57Z

Questions:

Should we add a sync.caught_up event for continuous datasets when they first reach chain head? I think this could be useful for consumers.
When a chain reorg causes progress to go backwards, should we:
- Emit a sync.reorg event?
- Suppress backwards progress events?
- Let consumers handle it (current behavior)?

cmwhited

I am going to keep reviewing. But I have concerns about the KafkaConfig
like I commented:

it is currently duplicated between the kafka client crate and the config crate
it doesn't expose all available connection params, including things like auth, which is concerning

crates/clients/kafka/proto/worker_events.proto

cmwhited · 2026-01-30T21:09:45Z

crates/clients/kafka/proto/worker_events.proto

+package amp.worker.events.v1;
+
+// Event envelope - common wrapper for all events
+message WorkerEvent {


So if my dataset has 3 tables, would each table emit this full WorkerEvent envelope? Seems like yes?
wonder if it would be less noise to make the payload an array for each tables status? so I could subscribe to a dataset stream and get the status of each table, to compute a total status.

Yes that's right.
So a 3 table dataset would emit 3x sync started events.
The partitioning is {namespace}/{name}/{hash}/{table_name} so consumers can subscribe to specific tables and get ordering guarantees per table.
Initially I had it so that I didn't have the lifecycle events following this process, but in the end I ended up deciding to include those as well.

crates/clients/kafka/src/config.rs

crates/config/src/lib.rs

crates/services/worker/src/events/kafka.rs

crates/services/worker/src/events.rs

crates/services/worker/src/events/types.rs

crates/services/worker/src/events/kafka.rs

LNSD

Please, check my comments 🙂

crates/clients/kafka/Cargo.toml

crates/clients/kafka/src/error.rs

crates/services/worker/build.rs

LNSD

Please, check my comments 🙂

crates/clients/kafka/proto/worker_events.proto

crates/core/dump/src/progress.rs

LNSD · 2026-02-03T11:29:14Z

As a piece of advice, this is a considerable PR (many logical changes and many files changed). Although it is ok from the POV of having an overall vision of the feature, I would recommend that you work on small incremental PR instead, as big PRs are prone to reworks due to merge conflicts.

Also, I recommend that you run the /code-review agent skill locally to catch code style issues in advance and accelerate the feedback loop.

…e-based

LNSD

Please, check my comments 🙂

LNSD · 2026-02-04T19:34:47Z

crates/services/worker/src/service.rs

    let notification_multiplexer = Arc::new(notification_multiplexer::spawn(metadata_db.clone()));

+    // Use injected event emitter or create one based on configuration
+    let event_emitter: Arc<dyn EventEmitter> = match event_emitter {


Can we make this more readable? More linear? 👀

https://github.com/zakirullin/cognitive-load#nested-ifs

Ok, I will add a helper function.

LNSD · 2026-02-04T19:38:59Z

tests/src/testlib/fixtures/mock_event_emitter.rs

+#[cfg(test)]
+mod tests {


The E2E tests are the tests for the fixtures. This adds maintenance overhead with little value, as the fixture is not production code.

Ok, no worries.
Good point. I will remove it.

LNSD · 2026-02-04T19:41:19Z

justfile

+[group: 'codegen']
+gen-worker-events-proto:
+    RUSTFLAGS="--cfg gen_worker_proto" cargo check -p worker


You should also add this to gen target

LNSD · 2026-02-04T19:44:24Z

tests/src/tests/it_worker_events.rs

+
+/// Test that completed events are recorded correctly.
+#[tokio::test]
+async fn test_completed_events_recorded() {


The test_* in the function name is redundant. We already know that this is a test 😅

The test function names should be descriptive to facilitate the test case navigation and the test coverage assessment.

Run the /code-review skill and tell Claude/Codex to focus on the test coding patterns. You can ask it to suggest better fn names and improvements based on the guidelines.

LNSD · 2026-02-04T19:46:21Z

tests/src/tests/it_worker_events.rs

+async fn test_progress_reporter_forwards_to_emitter() {
+    logging::init();
+
+    // Given


Tip/Nit: If you use //* Given instead of // Given, it helps to differentiate sections from comments

This is also stated in the coding guidelines.

mitchhs12 self-assigned this Jan 30, 2026

mitchhs12 force-pushed the worker-event-stream branch from 5c00f7a to 6942be7 Compare January 30, 2026 18:21

mitchhs12 closed this Jan 30, 2026

mitchhs12 reopened this Jan 30, 2026

mitchhs12 requested review from LNSD and cmwhited January 30, 2026 18:32

cmwhited requested changes Jan 30, 2026

View reviewed changes

cmwhited reviewed Jan 30, 2026

View reviewed changes

crates/services/worker/src/events/kafka.rs Outdated Show resolved Hide resolved

mitchhs12 force-pushed the worker-event-stream branch 2 times, most recently from 21c4346 to 0d4e966 Compare February 2, 2026 18:04

LNSD requested changes Feb 3, 2026

View reviewed changes

crates/clients/kafka/Cargo.toml Outdated Show resolved Hide resolved

crates/clients/kafka/src/error.rs Outdated Show resolved Hide resolved

crates/services/worker/build.rs Show resolved Hide resolved

LNSD requested changes Feb 3, 2026

View reviewed changes

crates/clients/kafka/proto/worker_events.proto Outdated Show resolved Hide resolved

crates/core/dump/src/progress.rs Outdated Show resolved Hide resolved

mitchhs12 force-pushed the worker-event-stream branch 4 times, most recently from b808ecf to 18552bf Compare February 4, 2026 19:59

mitchhs12 added 11 commits February 4, 2026 15:23

docs: update worker events for protobuf

6f8e784

WIP: feat(worker): add Kafka event streaming for sync progress

cb06837

feat(worker): emit lifecycle events per-table

fb796c4

feat(kafka): make partition count configurable

6716032

test(tests): fix E2E test to expect per-table events

611a1c5

chore(monitoring): add kafka_client to AMP_CRATES list

929a7a6

chore(deps): update Cargo.lock

944229f

refactor(worker): addresses PR feedback

08d62b5

fix(kafka): add error_source to error logs

6ab9dfa

refactor(worker): uses proto types instead for event streaming

152a2b5

refactor(kafka): consolidated kafka config into amp-config

0c157df

mitchhs12 added 8 commits February 4, 2026 15:26

docs(kafka): add status field and preqres section

412be35

refactor(worker): rename ProgressCallback to ProgressReporter

cb5799d

refactor(worker): change progress events from percentage-based to tim…

d64136e

…e-based

fix(dump): suppress clippy too_many_arguments warning

95b73ad

feat(worker): add SASL and TLS support for Kafka

2dd77ce

docs(worker-events): align feature doc with implementation

5ec25c4

add missing amp-config dep

6eb0a80

docs: fix config key and proto comment accuracy

c0da2c5

mitchhs12 force-pushed the worker-event-stream branch from 1ebcfb2 to c0da2c5 Compare February 4, 2026 20:27

mitchhs12 requested review from LNSD and cmwhited February 4, 2026 20:30

fix(worker): clippy

6b947be

LNSD reviewed Feb 4, 2026

View reviewed changes

		#[cfg(test)]
		mod tests {

Worker event stream #1665

Are you sure you want to change the base?

Worker event stream #1665

Conversation

mitchhs12 commented Jan 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Test plan

Uh oh!

mitchhs12 commented Jan 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cmwhited left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

LNSD left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

LNSD left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

LNSD commented Feb 3, 2026

Uh oh!

LNSD left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mitchhs12 commented Jan 30, 2026 •

edited

Loading

mitchhs12 commented Jan 30, 2026 •

edited

Loading