feat(taskbroker): Add Canary Task Sending to Taskbroker

[george-sentry]

Linear

Refs STREAM-1606

Description

As of taskbroker-client version 0.20.10, every taskworker application gets an internal namespace with canary_task baked in. It would be nice to validate that brand new pools work properly by automatically sending a handful of these tasks on startup rather than manually creating and then tearing down a task generator.

This is less rigorous than spawning a task generator as it doesn't tell us whether the consumer works, but it does tell us whether the components from the fetch threads onwards work, which is already good.

[linear-code]

STREAM-1606

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit b95c1a0. Configure here.}

[cursor]

Each replica enqueues duplicate canaries

Medium Severity

Every taskbroker process calls canary_tasks::enqueue on boot. With a shared Postgres activation store and horizontal scaling, each replica inserts another canary_tasks × worker_map batch with new UUIDs, so the same startup event can fan out far more internal canary work than the configured count intends.

Additional Locations (1)

• src/canary_tasks.rs#L27-L46

 

^{Reviewed by Cursor Bugbot for commit b95c1a0. Configure here.}

[george-sentry]

I'm okay with this for now, but this is another good reason to send canary tasks into a topic shared by the entire pool.

[sentry]

Bug: A transient database error when enqueuing canary tasks during startup will cause the entire broker to fail to start due to error propagation from canary_tasks::enqueue.
_{Severity: HIGH}

Suggested Fix

Instead of using the ? operator to propagate errors from canary_tasks::enqueue, handle the Result with a match statement or if let Err(...). Log the error for debugging purposes but allow the broker startup process to continue, as canary task insertion is not a critical startup step.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: src/main.rs#L113

Potential issue: The call to `canary_tasks::enqueue(&config, store.as_ref()).await?`
uses the `?` operator, which propagates any errors. A transient database error during
this operation will cause the `main` function to exit, preventing the entire broker from
starting. Since canary tasks are a non-critical diagnostic feature, a failure during
their insertion should not be a fatal startup error. This is inconsistent with how other
non-critical startup tasks, like `full_vacuum_on_start`, handle errors gracefully.

_{Did we get this right? 👍 / 👎 to inform future reviews.}