feat(stovepipe): tag controller metrics by queue

[mnoah1]

Summary

Tag Stovepipe controller and consumer-framework metrics with the Stovepipe queue that owns each message. Ingest places the queue name in context once; the publish framework persists it under the message metadata key queue_name and propagates it across later publishes. Before invoking the gate or controller, the consumer restores it to delivery context. metrics.TagsFromContext then emits it as the metric tag queue (for example, queue=monorepo/main).

Intent

• Make build, record, process, gate, acknowledgement, retry, and DLQ activity attributable to a queue.
• Make the queue available even for failures that happen before the payload can be decoded.
• Keep metric attribution request-local and safe when shared controllers process messages concurrently.
• Prevent individual pipeline stages from accidentally dropping queue attribution when publishing the next message.

Approach and reasoning

• Use message metadata instead of the partition value. A partition key controls transport ordering and is not guaranteed to represent the Stovepipe queue for every controller. queue_name states the intended meaning directly.
• Do not derive the tag from the payload. Payload queue fields remain authoritative for storage lookup and validation, but deserialization can fail before they are available. Metadata lets early consumer, gate, decode-error, and DLQ metrics still carry the queue.
• Extract metadata before invoking application code. The consumer copies only queue_name into a fresh delivery context before checking the gate or calling controller.Process.
• Propagate it in the publish framework. publish.Message and publish.MessageWithMetadata clone caller metadata and add the context queue when the caller did not supply that key. Stages no longer need to re-add queue metadata manually, while explicit metadata and intentional overrides remain supported.
• Keep context request-local. Per-delivery context avoids mutating shared controllers or Tally scopes, so concurrent messages cannot overwrite one another’s queue attribution.
• Allowlist context-derived data. Other message metadata has separate semantics, such as failure details. Only the queue name enters context or automatic metric tags.
• Keep the names clear across layers. The serialized metadata key is queue_name, code accesses it through messagequeue.QueueName(ctx), and the resulting metric dimension is named queue.
• Centralize metric extraction. metrics.TagsFromContext defines which context values may become tags; it currently extracts only queue.
• Remain rollout-safe. Messages published before queue_name was introduced simply emit metrics without the queue tag.

Changes

• Seed queue context at the initial Stovepipe ingest handoff.
• Automatically persist and propagate it as queue_name metadata through both platform publish helpers.
• Copy only queue_name into delivery context before gate and controller processing.
• Add queue-specific context accessors and metrics.TagsFromContext.
• Apply the queue metric tag to primary controller, DLQ controller, gate, and message outcome metrics.
• Cover propagation, metadata merging and immutability, explicit overrides, missing metadata, and emitted queue-tagged series.

Test Plan

• ./tool/bazel test //platform/base/messagequeue:go_default_test //platform/publish:go_default_test //platform/consumer:go_default_test //platform/metrics:go_default_test //stovepipe/controller:go_default_test //stovepipe/controller/build:go_default_test //stovepipe/controller/buildsignal:go_default_test //stovepipe/controller/dlq:go_default_test //stovepipe/controller/process:go_default_test //stovepipe/controller/record:go_default_test
• Deploy and verify affected metric series contain queue=<queue-name> while older messages without metadata continue processing normally.

Revert Plan

• Revert the commit to remove queue metadata propagation and restore the prior untagged metrics.

Issues

• None.

[behinddwalls]

i am wondering if this should depend the base messagequeue context or not? this is bit central in nature so if we add this and say we need something else in some other part of the platform would we also put it here? can we pass the keys which we need to extract in the call itself so we don't need to depend on messagequeue entity itself