From f3a8a144d42862aec9d22a0f471a4698c57a0b97 Mon Sep 17 00:00:00 2001 From: "prath.shenoy" Date: Thu, 5 Nov 2026 19:39:46 +0000 Subject: [PATCH] fix(stovepipe): Track validations in order --- doc/rfc/stovepipe/steps/build.md | 2 +- doc/rfc/stovepipe/steps/buildanalysis.md | 4 +- doc/rfc/stovepipe/steps/process.md | 24 +- platform/extension/messagequeue/README.md | 2 + .../messagequeue/subscription_config.go | 8 + .../messagequeue/subscription_config_test.go | 7 + service/stovepipe/server/main.go | 4 +- service/stovepipe/server/main_test.go | 1 + .../controller/build/build_analysis_test.go | 16 +- .../controller/buildanalysis/BUILD.bazel | 3 - stovepipe/controller/buildanalysis/README.md | 18 +- .../controller/buildanalysis/buildanalysis.go | 46 +-- .../buildanalysis/buildanalysis_test.go | 121 +------- .../controller/buildanalysis/history_test.go | 31 +- stovepipe/controller/issuetrack/README.md | 32 +- .../controller/issuetrack/delivery_test.go | 287 +++++++++++++++--- stovepipe/controller/issuetrack/dlq.go | 98 +++--- stovepipe/controller/issuetrack/dlq_test.go | 99 ++++-- .../controller/issuetrack/fix_handoff_test.go | 45 ++- .../controller/issuetrack/fixture_test.go | 16 +- stovepipe/controller/issuetrack/history.go | 16 + .../controller/issuetrack/history_test.go | 8 +- stovepipe/controller/issuetrack/issuetrack.go | 156 +++++++--- stovepipe/controller/process/process.go | 37 ++- stovepipe/controller/process/process_test.go | 121 ++++++-- .../core/messagequeue/messagequeue_test.go | 30 +- .../core/messagequeue/proto/issuetrack.proto | 17 +- .../messagequeue/protopb/issuetrack.pb.go | 39 +-- stovepipe/entity/issue_tracking_state.go | 22 +- stovepipe/entity/issue_tracking_state_test.go | 19 ++ stovepipe/extension/storage/README.md | 4 +- test/e2e/stovepipe/issuetrack_test.go | 10 +- .../messagequeue/mysql/queue_test.go | 28 ++ 33 files changed, 884 insertions(+), 487 deletions(-) diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md index c5c36349f..0c483eecc 100644 --- a/doc/rfc/stovepipe/steps/build.md +++ b/doc/rfc/stovepipe/steps/build.md @@ -1,6 +1,6 @@ # Build stage -Build dispatch retains the request's build association before publishing status and analysis notices. Retries reuse that association instead of triggering again; an uncertain build create is accepted only when its retained request identity matches. +Build dispatch retains the request's build association before publishing status and analysis notices. Retries reuse that association instead of triggering again; an uncertain build create is accepted only when its retained request identity matches. This supplies the build ID to the request-scoped tracking job already registered at admission. `build` triggers the build-runner for the scope `process` already decided and hands the resulting build id to `buildsignal`. See [workflow.md](../workflow.md) for where it sits in the pipeline, and [process.md](process.md) for how the scope it reads (`BuildStrategy`, `BaseURI`) is chosen. diff --git a/doc/rfc/stovepipe/steps/buildanalysis.md b/doc/rfc/stovepipe/steps/buildanalysis.md index f80765a15..31c270007 100644 --- a/doc/rfc/stovepipe/steps/buildanalysis.md +++ b/doc/rfc/stovepipe/steps/buildanalysis.md @@ -1,6 +1,6 @@ # Build issue detection -Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success. The reference host consumes issue-tracking notifications and registers their DLQ recovery alongside analysis. +Build analysis runs alongside status polling and recording. Hosts supply a queue-resolved detector factory. The reference service uses a no-op detector so the analysis stage is always wired without inspecting provider output; deployments replace the factory with their integration. No-op results do not establish build success. The reference host consumes request-scoped issue-tracking jobs and registers their DLQ recovery alongside analysis. ## Start and completion notices @@ -12,7 +12,7 @@ Publication errors are returned through the producing stage's retry/DLQ policy. ## Build issues and polling -The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build issue persistence, polling, and the stop condition for Requests that end without a recorded build result. It accumulates per-build issues and publishes queue-partitioned issue-tracking notifications after persistence; it does not maintain open issues itself. Successful terminal analysis requests final reconciliation only after the Request records that Build as its winning outcome. Terminal detection must return the complete cumulative issue set, including issues returned by earlier calls. Analysis persists that set and retains `build_analysis_completed` before publishing final reconciliation. The first retained completion snapshots the issue count and timestamp; retries reuse that snapshot while still persisting detected issues and publishing the handoff. A changed detector count is logged rather than conflicting with history. Nonterminal polls and terminal passes waiting for the recorded winning outcome add no completion event. Issue tracking applies all accumulated build issues before permitting any absence-based closure; a task-wide failure preserves existing issues across all tasks. See the controller documentation for handoff and deduplication rules. +The [analysis controller](../../../../stovepipe/controller/buildanalysis/README.md) owns detection, per-build issue persistence, polling, and the stop condition for Requests that end without a recorded build result. It accumulates per-build issues; it does not maintain open issues itself. Admission registers one tracking job in request order before build dispatch. That job remains current through refreshes and waits for durable terminal readiness, while detection for newer validations continues independently. Successful terminal analysis requests final reconciliation only after the Request records that Build as its winning outcome. Terminal detection must return the complete cumulative issue set, including issues returned by earlier calls. Analysis persists that set and retains `build_analysis_completed` to authorize final reconciliation independently of completion-message delivery. The first retained completion snapshots the issue count and timestamp; retries reuse that snapshot while still persisting detected issues. A changed detector count is logged rather than conflicting with history. Nonterminal polls and terminal passes waiting for the recorded winning outcome add no completion event. Issue tracking applies all accumulated build issues before permitting any absence-based closure; a task-wide failure preserves existing issues across all tasks. See the controller documentation for handoff and deduplication rules. Status-polling DLQ recovery does not publish a build-completion notice. The analysis polling loop observes the ended Request through storage instead. diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index 64b941f59..24e9fffa9 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/doc/rfc/stovepipe/steps/process.md @@ -21,7 +21,7 @@ For a delivery carrying request id `R`: 2. If R.State is terminal (superseded / succeeded / failed / cancelled): - ack and return (idempotent no-op). 3. If R.State is processing (strategy already recorded): - - re-announce validation start and re-publish R to build (either prior publish may have failed), ack, return. + - re-announce validation start and re-publish R to build, ack, return. 4. R.State is accepted. Load the Queue row Q. 5. Coalesce: if CompareRequestID(R.Queue, R.ID, Q.latest_request_id) < 0: - a newer head exists -> mark R superseded, ack, return. (No slot consumed.) @@ -29,15 +29,19 @@ For a delivery carrying request id `R`: - defer (hold the delivery) -> re-check on redelivery until the slot frees (admit) or a newer head supersedes it. See [Waiting for a slot](#waiting-for-a-slot). 7. Admit R: a. Derive build strategy + baseline (see "Build-strategy decision"). - b. CAS the Request: accepted -> processing, persist build_strategy + base_uri. + b. Register the tracking job, then CAS the Queue version to revalidate this head. + - version conflict -> hold for immediate redelivery, rechecking the head and gate from step 1. + c. CAS the Request: accepted -> processing, persist build_strategy + base_uri. - lost to another writer (no longer accepted) -> ack; that writer owns the outcome and nothing is counted. - c. CAS the Queue row: in_flight_count += 1, reloading and retrying on a version conflict without re-gating. - d. Announce validation start on the hook topic (see "Hooks"). - e. Publish R to build. - f. ack. + d. CAS the Queue row: in_flight_count += 1, reloading and retrying on a version conflict without re-gating. + e. Announce validation start on the hook topic (see "Hooks"). + f. Publish R to build. + g. ack. ``` -Step 5 runs regardless of the gate: an intermediate head is superseded on sight (even mid-validation), because superseding consumes no slot. +Step 5 runs before the gate: an accepted intermediate head is superseded even when all build slots are occupied, because superseding consumes no slot. + +Step 7b reserves the request's place in [issue tracking](../../../../stovepipe/controller/issuetrack/README.md#validation-order) before its build can start. The Queue-version CAS checks that no newer head overtook the registration. If the request is superseded before admission, its tracking job is acknowledged without updating issues. ## Build-strategy decision @@ -149,7 +153,7 @@ The event type is `validation.repository.started`. Its payload names the Queue a Published after the admit CAS and before the publish to `build`: ``` -CAS accepted -> processing → publish HookEvent → publish to build → ack +register tracking → validate Queue version → CAS accepted -> processing → publish HookEvent → publish to build → ack ``` After the CAS because the payload names the Request rather than snapshotting it, and the two facts a start event exists to carry — the scope it chose and the baseline it builds on — are written by that very CAS. A hook that reloads the Request must not find it still `accepted` with neither set. Before the build publish because the announce is the cheaper of the two to retry: a failed announce leaves nothing downstream to undo, whereas announcing after the build publish would make a failed announce force the redelivery to re-publish a build that was already accepted. @@ -160,8 +164,8 @@ Only an admit announces. A Request that coalescing supersedes never reaches step Every branch is safe under redelivery: -- **accepted, no strategy** → full admit path. Nothing has been written yet: a crash before the request CAS leaves no trace, and redelivery re-runs the gate and the admit from scratch. -- **processing** → re-announce the start event, re-publish to `build`, ack. The `build` consumer is keyed on the request id and idempotent, so a duplicate publish is harmless, and the start event's id is derived from the transition rather than the clock, so a re-announce carries the id the first attempt would have and consumers dedupe on it. Re-announcing here is what makes the event at-least-once rather than at-most-once: this is the only branch a redelivery takes once `processing` is durable, so an admit that failed after the state write would otherwise lose the event for good. +- **accepted, no strategy** → full admit path. A crash before the request CAS may leave a tracking registration, but no admitted request. Redelivery re-runs the gate and admission using the same registration identity. +- **processing** → re-announce the start event, re-publish to `build`, ack. Tracking registration already succeeded before the processing transition, so it is not repeated. Stable message IDs make repeated hook and build publications idempotent. Re-announcing repairs an admission that persisted the state but failed before publication. - **terminal** (superseded / recorded) → ack, no-op. - **deferred (waiting for slot)** → no state or count change; pure deferral (re-enters when the held delivery comes due). diff --git a/platform/extension/messagequeue/README.md b/platform/extension/messagequeue/README.md index 1dd73f841..c6ee4da7c 100644 --- a/platform/extension/messagequeue/README.md +++ b/platform/extension/messagequeue/README.md @@ -72,6 +72,8 @@ See `subscription_config.go` for all fields and defaults. `Retry.MaxAttempts` uses zero to mean unlimited attempts. `ExtendedSubscriptionConfig` provides 10 attempts with a longer capped exponential backoff for idempotent stages that should tolerate transient dependency failures before dead-lettering. `DLQSubscriptionConfig` selects unlimited attempts and disables a second-level DLQ so reconciliation messages remain retryable until they converge or an operator removes them. +`SingleMessageSubscriptionConfig` uses extended retries and fetches at most one message per poll. Use it when holding a message must prevent prefetching later messages in its partition. + ## Usage ```go diff --git a/platform/extension/messagequeue/subscription_config.go b/platform/extension/messagequeue/subscription_config.go index 8ec839906..db5da486c 100644 --- a/platform/extension/messagequeue/subscription_config.go +++ b/platform/extension/messagequeue/subscription_config.go @@ -113,6 +113,14 @@ func ExtendedSubscriptionConfig(subscriberName, consumerGroup string) Subscripti return config } +// SingleMessageSubscriptionConfig uses extended retries and fetches at most one message per poll. +// Use it when holding a message must prevent prefetching later messages in its partition. +func SingleMessageSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig { + config := ExtendedSubscriptionConfig(subscriberName, consumerGroup) + config.BatchSize = 1 + return config +} + // DefaultSubscriptionConfig returns a SubscriptionConfig with sensible defaults. func DefaultSubscriptionConfig(subscriberName, consumerGroup string) SubscriptionConfig { return SubscriptionConfig{ diff --git a/platform/extension/messagequeue/subscription_config_test.go b/platform/extension/messagequeue/subscription_config_test.go index 175971eff..85ddace12 100644 --- a/platform/extension/messagequeue/subscription_config_test.go +++ b/platform/extension/messagequeue/subscription_config_test.go @@ -97,6 +97,13 @@ func TestExtendedSubscriptionConfig(t *testing.T) { assert.True(t, config.DLQ.Enabled) } +func TestSingleMessageSubscriptionConfig(t *testing.T) { + expected := ExtendedSubscriptionConfig("worker-1", "consumer-1") + expected.BatchSize = 1 + + assert.Equal(t, expected, SingleMessageSubscriptionConfig("worker-1", "consumer-1")) +} + func TestSubscriptionConfig_DifferentConsumerGroups(t *testing.T) { // Test that different consumer groups get independent configs tests := []struct { diff --git a/service/stovepipe/server/main.go b/service/stovepipe/server/main.go index d57fbce17..9d388c2df 100644 --- a/service/stovepipe/server/main.go +++ b/service/stovepipe/server/main.go @@ -417,7 +417,7 @@ func registerPrimaryControllers( } count++ - analysisController := buildanalysis.NewController(logger, scope, store, materializer, detectorFactory, registry, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis") + analysisController := buildanalysis.NewController(logger, scope, store, materializer, detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis") if err := c.Register(analysisController); err != nil { return count, fmt.Errorf("failed to register buildanalysis controller: %w", err) } @@ -598,7 +598,7 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe Key: stovepipemq.TopicKeyIssueTrack, Name: "issuetrack", Queue: q, - Subscription: extqueue.ExtendedSubscriptionConfig(subscriberName, "stovepipe-issuetrack"), + Subscription: extqueue.SingleMessageSubscriptionConfig(subscriberName, "stovepipe-issuetrack"), }, { Key: dlq.TopicKey(stovepipemq.TopicKeyIssueTrack), diff --git a/service/stovepipe/server/main_test.go b/service/stovepipe/server/main_test.go index a27c62e7c..abea3bca0 100644 --- a/service/stovepipe/server/main_test.go +++ b/service/stovepipe/server/main_test.go @@ -112,6 +112,7 @@ func TestIssueTrackingRegistration(t *testing.T) { cfg, ok := registry.SubscriptionConfig(stovepipemq.TopicKeyIssueTrack, "stovepipe-issuetrack") require.True(t, ok) + assert.Equal(t, 1, cfg.BatchSize) assert.True(t, cfg.DLQ.Enabled) assert.Equal(t, "_dlq", cfg.DLQ.TopicSuffix) } diff --git a/stovepipe/controller/build/build_analysis_test.go b/stovepipe/controller/build/build_analysis_test.go index c2720c73e..6b296c2a8 100644 --- a/stovepipe/controller/build/build_analysis_test.go +++ b/stovepipe/controller/build/build_analysis_test.go @@ -143,7 +143,7 @@ func TestBuildAnalysisPipeline(t *testing.T) { ) analysisController := buildanalysis.NewController(zap.NewNop().Sugar(), tally.NoopScope, - staticStorageFactory{store: m.store}, requestlog.NewMaterializer(tally.NoopScope), detectorFactory, buildController.registry, stovepipemq.TopicKeyBuildAnalysis, "analysis") + staticStorageFactory{store: m.store}, requestlog.NewMaterializer(tally.NoopScope), detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "analysis") statusController := buildsignal.NewController(zap.NewNop().Sugar(), tally.NoopScope, staticStorageFactory{store: m.store}, m.materializer, m.runnerFactory, buildController.registry, stovepipemq.TopicKeyBuildSignal, "signal") @@ -184,19 +184,7 @@ func TestBuildAnalysisPipeline(t *testing.T) { assert.True(t, persisted[b]) assert.True(t, persisted[a]) - require.Len(t, published["issuetrack"], 4) - for i, msg := range published["issuetrack"] { - assert.Equal(t, testQueue, msg.PartitionKey) - assert.Equal(t, testQueue, msg.Tenant) - notification := &stovepipemq.IssueTrack{} - require.NoError(t, stovepipemq.Unmarshal(msg.Payload, notification)) - assert.Equal(t, testBuildID, notification.GetBuildId()) - assert.Equal(t, testQueue, notification.GetQueueName()) - assert.Equal(t, i > 1, notification.GetFinalize()) - } - assert.NotEqual(t, published["issuetrack"][0].ID, published["issuetrack"][1].ID) - assert.NotEqual(t, published["issuetrack"][1].ID, published["issuetrack"][2].ID) - assert.Equal(t, published["issuetrack"][2].ID, published["issuetrack"][3].ID) + require.Empty(t, published["issuetrack"]) require.Len(t, history, 1) for _, log := range history { assert.Equal(t, entity.RequestEventBuildAnalysisCompleted, log.Event) diff --git a/stovepipe/controller/buildanalysis/BUILD.bazel b/stovepipe/controller/buildanalysis/BUILD.bazel index 39f5ea62d..b70f9b591 100644 --- a/stovepipe/controller/buildanalysis/BUILD.bazel +++ b/stovepipe/controller/buildanalysis/BUILD.bazel @@ -12,7 +12,6 @@ go_library( "//platform/base/messagequeue:go_default_library", "//platform/consumer:go_default_library", "//platform/metrics:go_default_library", - "//platform/publish:go_default_library", "//stovepipe/core/loader:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/requestlog:go_default_library", @@ -38,9 +37,7 @@ go_test( "//platform/consumer:go_default_library", "//platform/consumer/mock:go_default_library", "//platform/errs:go_default_library", - "//platform/extension/messagequeue/mock:go_default_library", "//platform/metrics:go_default_library", - "//platform/publish:go_default_library", "//stovepipe/core/messagequeue:go_default_library", "//stovepipe/core/requestlog:go_default_library", "//stovepipe/core/requestlog/mock:go_default_library", diff --git a/stovepipe/controller/buildanalysis/README.md b/stovepipe/controller/buildanalysis/README.md index 1e1a65bd3..4b6aec4aa 100644 --- a/stovepipe/controller/buildanalysis/README.md +++ b/stovepipe/controller/buildanalysis/README.md @@ -4,29 +4,29 @@ The controller reloads the queue-scoped Build, invokes the queue's issue detector, and persists each returned build issue individually. Build issues already stored for the same build, task, and unit within a queue are not persisted again, so repeated polls and retries after partial writes are safe. Detector results accompanying an error are discarded. -Once the Request records the terminal Build as its winning outcome, successful detection and issue persistence retain `build_analysis_completed` before the final issue-tracking handoff. The event snapshots the build ID and issue count, including zero, from the first retained completion. Retries still detect and persist issues, but reuse the retained count and timestamp; a changed count is logged as a detector discrepancy and does not block publication. Concurrent deliveries and uncertain writes converge on the first stored snapshot, while other identity conflicts remain errors. Nonterminal polls and terminal passes waiting for the winning outcome add no completion entry. +Once the Request records the terminal Build as its winning outcome, successful detection and issue persistence retain `build_analysis_completed` to authorize final reconciliation. The event snapshots the build ID and issue count, including zero, from the first retained completion. Retries still detect and persist issues, but reuse the retained count and timestamp; a changed count is logged as a detector discrepancy and does not block completion. Concurrent deliveries and uncertain writes converge on the first stored snapshot, while other identity conflicts remain errors. Nonterminal polls and terminal passes waiting for the winning outcome add no completion entry. -After persistence, the controller publishes an issue-tracking notification partitioned by queue. Issue tracking maintains open issues and lifecycle history; analysis never writes them directly. An empty detection result does not delete earlier build issues. +Admission registers a queue-partitioned tracking job for each validation before build dispatch. After persistence, analysis retains terminal completion in request history; the tracking job reads that event independently of queue delivery. Issue tracking maintains open issues and lifecycle history; analysis never writes them directly. An empty detection result does not delete earlier build issues. ## Issue-tracking handoff -Refresh notifications let issue tracking open issues reported by the build without closing other open issues. Final notifications let it also close open issues absent from the build's accumulated issues, unless any task-wide failure was recorded or the build was canceled. A successful terminal detection must return the complete cumulative issue set for the build, including issues returned by earlier calls. Persisting that complete result before publication makes finalization independent of whether an earlier worker finished saving its results. An empty terminal result is valid only if the build has no reportable issues; inability to determine the complete set must return an error instead. See the [detector contract](../../extension/buildissue/README.md). +The request-scoped tracking job reads persisted build issues during ordinary refreshes without closing other issues. Once analysis records completion, tracking can also close issues absent from the accumulated build issues, unless any task-wide failure was recorded or the build was canceled. Analysis does not publish tracking messages. -All analysis deliveries for a build must share its build-ID partition and consumer group. The controller validates that partition key. Finalization requires a successful pass that began with a terminal Build and the Request's recorded winning build ID. If status becomes terminal before that outcome is recorded, analysis publishes a refresh and holds until the outcome is available. Known losing builds do not publish. A terminal Request without a recorded winning build permits a refresh but not closing issues based on absence. +A successful terminal detection must return the complete cumulative issue set for the build, including issues returned by earlier calls. Persisting that result before recording completion makes finalization independent of whether an earlier worker finished saving its results. An empty terminal result is valid only if the build has no reportable issues; inability to determine the complete set must return an error instead. See the [detector contract](../../extension/buildissue/README.md). -Every refresh gets a fresh notification ID because held polling deliveries do not identify individual passes. A fixed ID would suppress later refreshes after earlier notifications were consumed. Retries can therefore publish duplicate refreshes; issue tracking is idempotent. Final notifications use one deterministic ID per build, so duplicate terminal passes deduplicate. Publication errors return through the normal analysis retry/DLQ policy; no hold or successful completion occurs before publication succeeds. +All analysis deliveries for a build share its build-ID partition and consumer group. Completion requires a successful pass that began with a terminal Build and the Request's recorded winning build ID. If status becomes terminal before that outcome is recorded, analysis persists the issues and holds until the outcome is available. Known losing builds stop without detection. A Request that ends without a winning build cannot authorize closing issues based on absence; tracking abandons that validation through its normal recovery. ## Polling and completion -Every delivery reloads the Build and its Request before detection. After a successful pass, a nonterminal Build holds the delivery for two minutes; a terminal Build completes after persistence and publication once its Request outcome is known, or the Request has ended without one. A pass that begins with a nonterminal Build always holds, even if the Build finishes during detection, so another delivery analyzes it after observing terminal status. More than one delivery may analyze the same terminal Build; build-issue persistence remains idempotent. +Every delivery reloads the Build and its Request before detection. After a successful pass, a nonterminal Build holds the delivery for two minutes; a terminal Build completes after persistence and completion history once its Request outcome is known, or the Request has ended without one. A pass that begins with a nonterminal Build always holds, even if the Build finishes during detection, so another delivery analyzes it after observing terminal status. More than one delivery may analyze the same terminal Build; build-issue persistence remains idempotent. -If the Request is terminal but the loaded Build is nonterminal, analysis reloads the Build before deciding to stop. The Build may have finished between the initial Build read and the Request read; reloading avoids skipping its final analysis. If it is still nonterminal, analysis stops without detection or a hold. This can occur when the status-polling DLQ handler fails the Request without learning the actual build result, leaving the Build at its last observed status. +If the Request is terminal but the loaded Build is nonterminal, analysis reloads the Build before deciding to stop. The Build may have finished between the initial Build read and the Request read; reloading avoids skipping its final analysis. If it is still nonterminal, analysis retains abandonment and stops without detection or a hold. This can occur when the status-polling DLQ handler fails the Request without learning the actual build result, leaving the Build at its last observed status. Status polling and updates remain the responsibility of buildsignal. Analysis never changes Request state or releases build slots. ## Errors -Detection, persistence, publication, configuration, state-loading, and message-validation errors are returned to the consumer for its normal retry or DLQ policy, whether the Build is running or terminal. Analysis errors do not use the two-minute polling hold. The consumer classifies errors and applies its configured retry limits and backoff; a final analysis can still fail and dead-letter. +Detection, persistence, configuration, state-loading, and message-validation errors are returned to the consumer for its normal retry or DLQ policy, whether the Build is running or terminal. Analysis errors do not use the two-minute polling hold. The consumer classifies errors and applies its configured retry limits and backoff; a final analysis can still fail and dead-letter. ## DLQ recovery @@ -38,4 +38,4 @@ State-read and history-write errors are returned for retry. Hosts must run recov ## Host integration -The reference service supplies a no-op detector and publishes analysis notices at build start and completion. It registers both analysis and issue tracking with bounded normal retries and the existing DLQ consumer for recovery. See [issue tracking](../issuetrack/README.md) for reconciliation and recovery behavior and [build issue detection](../../../doc/rfc/stovepipe/steps/buildanalysis.md) for the publication rules and retry configuration. +The reference service supplies a no-op detector, registers tracking at admission, and publishes analysis notices at build start and completion. It registers both analysis and issue tracking with bounded normal retries and the existing DLQ consumer for recovery. See [issue tracking](../issuetrack/README.md) for reconciliation and recovery behavior and [build issue detection](../../../doc/rfc/stovepipe/steps/buildanalysis.md) for the publication rules and retry configuration. diff --git a/stovepipe/controller/buildanalysis/buildanalysis.go b/stovepipe/controller/buildanalysis/buildanalysis.go index 01a46bcd0..eaa448ad3 100644 --- a/stovepipe/controller/buildanalysis/buildanalysis.go +++ b/stovepipe/controller/buildanalysis/buildanalysis.go @@ -26,7 +26,6 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/platform/metrics" - "github.com/uber/submitqueue/platform/publish" "github.com/uber/submitqueue/stovepipe/core/loader" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" @@ -41,14 +40,13 @@ import ( const PollDelayMs int64 = 120000 // Controller accumulates build issues while builds run and after completion. -// Terminal analysis must finish persistence and publication before its delivery completes. +// Terminal analysis must persist all build issues and its completion event before acknowledging. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope stores storage.Factory materializer requestlog.Materializer detectorFactory buildissue.DetectorFactory - registry consumer.TopicRegistry topicKey consumer.TopicKey consumerGroup string } @@ -59,7 +57,6 @@ func NewController( stores storage.Factory, materializer requestlog.Materializer, detectorFactory buildissue.DetectorFactory, - registry consumer.TopicRegistry, topicKey consumer.TopicKey, consumerGroup string, ) *Controller { @@ -69,13 +66,12 @@ func NewController( stores: stores, materializer: materializer, detectorFactory: detectorFactory, - registry: registry, topicKey: topicKey, consumerGroup: consumerGroup, } } -// Process persists before publication and holds while further analysis is needed. +// Process persists detected issues and holds while further analysis is needed. // Errors use normal queue retry or DLQ handling; partial writes are safe to retry. func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { op := metrics.Begin(c.metricsScope, "process", metrics.LongLatencyBuckets, metrics.TagsFromContext(ctx)...) @@ -131,7 +127,9 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r if !build.Status.IsTerminal() { c.logger.Warnw("stopping analysis for a terminal request with nonterminal build", "queue", request.Queue, "build_id", build.ID, "request_id", request.ID, "request_state", string(request.State)) - return nil + log := requestlog.NewRequestEventLog(request, entity.RequestEventBuildAnalysisAbandoned, build.ID, + map[string]string{requestlog.MetadataKeyBuildID: build.ID}) + return c.materializer.PersistLog(ctx, store, log) } } @@ -153,7 +151,7 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r } hold := !build.Status.IsTerminal() - finalize := false + complete := false if build.Status.IsTerminal() { // Build status can become terminal before buildsignal records the winning build. request, err = loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request") @@ -166,18 +164,15 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID { return nil } - finalize = request.TerminalBuildID == build.ID - hold = !finalize && !request.State.IsTerminal() + complete = request.TerminalBuildID == build.ID + hold = !complete && !request.State.IsTerminal() } - if finalize { + if complete { if err := c.retainCompletedAnalysis(ctx, store, request, build.ID, len(issues)); err != nil { return fmt.Errorf("retain completed analysis for build %q: %w", build.ID, err) } - } - - if err := c.publishIssueTracking(ctx, request.Queue, build.ID, finalize); err != nil { - return err + c.logger.Infow("terminal issue analysis ready", "queue", request.Queue, "request_id", request.ID, "build_id", build.ID, "issue_count", len(issues)) } if hold { @@ -232,27 +227,6 @@ func (c *Controller) retainCompletedAnalysis(ctx context.Context, store storage. return c.materializer.PersistLog(ctx, store, log) } -func (c *Controller) publishIssueTracking(ctx context.Context, queue, buildID string, finalize bool) error { - payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{ - BuildId: buildID, QueueName: queue, Finalize: finalize, - }) - if err != nil { - return fmt.Errorf("serialize issue tracking notification: %w", err) - } - - // Polls reuse one delivery, so no stable occurrence ID distinguishes refreshes. - messageID := publish.UniqueID(buildID) - if finalize { - messageID = publish.IntentID(buildID, "finished") - } - if err := publish.Message(ctx, c.registry, stovepipemq.TopicKeyIssueTrack, publish.MessageParams{ - Tenant: queue, ID: messageID, Payload: payload, PartitionKey: queue, - }); err != nil { - return fmt.Errorf("publish issue tracking notification: %w", err) - } - return nil -} - func (c *Controller) Name() string { return "buildanalysis" } func (c *Controller) TopicKey() consumer.TopicKey { return c.topicKey } diff --git a/stovepipe/controller/buildanalysis/buildanalysis_test.go b/stovepipe/controller/buildanalysis/buildanalysis_test.go index 35d45cdcc..e542026fd 100644 --- a/stovepipe/controller/buildanalysis/buildanalysis_test.go +++ b/stovepipe/controller/buildanalysis/buildanalysis_test.go @@ -23,12 +23,9 @@ import ( "github.com/stretchr/testify/require" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" - "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" "github.com/uber/submitqueue/platform/errs" - mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock" "github.com/uber/submitqueue/platform/metrics" - "github.com/uber/submitqueue/platform/publish" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" @@ -62,7 +59,6 @@ type analysisMocks struct { issues *storagemock.MockBuildIssueStore detectorFactory *buildissuemock.MockDetectorFactory detector *buildissuemock.MockDetector - publisher *mqmock.MockPublisher scope tally.TestScope } @@ -80,7 +76,6 @@ func newTestController(t *testing.T) (*Controller, analysisMocks) { issues: storagemock.NewMockBuildIssueStore(ctrl), detectorFactory: buildissuemock.NewMockDetectorFactory(ctrl), detector: buildissuemock.NewMockDetector(ctrl), - publisher: mqmock.NewMockPublisher(ctrl), scope: tally.NewTestScope("test", nil), } @@ -112,14 +107,7 @@ func newTestController(t *testing.T) (*Controller, analysisMocks) { } return log, nil }).AnyTimes() - queue := mqmock.NewMockQueue(ctrl) - queue.EXPECT().Publisher().Return(m.publisher).AnyTimes() - registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ - {Key: stovepipemq.TopicKeyIssueTrack, Name: "issue_track", Queue: queue}, - }) - require.NoError(t, err) - - c := NewController(zap.NewNop().Sugar(), m.scope, m.stores, m.materializer, m.detectorFactory, registry, + c := NewController(zap.NewNop().Sugar(), m.scope, m.stores, m.materializer, m.detectorFactory, stovepipemq.TopicKeyBuildAnalysis, "stovepipe-buildanalysis") return c, m @@ -155,31 +143,6 @@ func (m analysisMocks) expectPass(status entity.BuildStatus, issues []entity.Bui return m.detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(issues, err).After(detector) } -func (m analysisMocks) expectNotification(t *testing.T, finalize bool, after *gomock.Call) *gomock.Call { - t.Helper() - call := m.publisher.EXPECT().Publish(gomock.Any(), "issue_track", gomock.Any()). - DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { - assert.Equal(t, testQueue, msg.Tenant) - assert.Equal(t, testQueue, msg.PartitionKey) - assert.Equal(t, testQueue, msg.Metadata[entityqueue.MetadataKeyQueueName]) - notification := &stovepipemq.IssueTrack{} - require.NoError(t, stovepipemq.Unmarshal(msg.Payload, notification)) - assert.Equal(t, testBuildID, notification.GetBuildId()) - assert.Equal(t, testQueue, notification.GetQueueName()) - assert.Equal(t, finalize, notification.GetFinalize()) - if finalize { - assert.Equal(t, publish.IntentID(testBuildID, "finished"), msg.ID) - } else { - assert.NotEmpty(t, msg.ID) - } - return nil - }) - if after != nil { - call.After(after) - } - return call -} - func TestProcessPersistsBeforeHoldingOrCompleting(t *testing.T) { issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit-a"} @@ -199,10 +162,12 @@ func TestProcessPersistsBeforeHoldingOrCompleting(t *testing.T) { if !empty { last = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(last) } - last = m.expectNotification(t, status.IsTerminal(), last) d := m.delivery(t) - if !status.IsTerminal() { + if status.IsTerminal() { + m.historyRead.After(last) + m.historyCreate.After(last) + } else { d.EXPECT().Hold(PollDelayMs).After(last) } @@ -225,7 +190,6 @@ func TestProcessAcceptsRepeatedAndTaskWideObservations(t *testing.T) { last = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(last) last = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue). Return(fmt.Errorf("already recorded: %w", storage.ErrAlreadyExists)).After(last) - last = m.expectNotification(t, false, last) d := m.delivery(t) d.EXPECT().Hold(PollDelayMs).After(last) @@ -259,7 +223,6 @@ func TestProcessRetriesPartialPersistenceAndReloadsStatus(t *testing.T) { require.ErrorIs(t, err, writeErr) assert.True(t, errs.IsRetryable(err)) } else { - last = m.expectNotification(t, pass.status.IsTerminal(), last) if !pass.status.IsTerminal() { d.EXPECT().Hold(PollDelayMs).After(last) } @@ -290,13 +253,11 @@ func TestProcessRunsAgainWhenBuildCompletesDuringDetection(t *testing.T) { first := m.delivery(t) first.EXPECT().Hold(PollDelayMs) - m.expectNotification(t, false, nil) require.NoError(t, c.Process(context.Background(), first)) m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil) m.request.TerminalBuildID = testBuildID - m.expectNotification(t, true, nil) require.NoError(t, c.Process(context.Background(), m.delivery(t))) } @@ -440,8 +401,7 @@ func TestProcessStillAnalyzesTerminalBuildOfTerminalRequest(t *testing.T) { issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit"} detected := m.expectPass(entity.BuildStatusFailed, []entity.BuildIssue{issue}, nil) - persisted := m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected) - m.expectNotification(t, true, persisted) + m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected) require.NoError(t, c.Process(context.Background(), m.delivery(t))) } @@ -470,8 +430,7 @@ func TestCompletionBetweenBuildAndRequestReadStillAnalyzes(t *testing.T) { m.detectorFactory.EXPECT().For(buildissue.DetectorConfig{QueueName: testQueue}).Return(m.detector, nil) issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit"} detected := m.detector.EXPECT().Detect(gomock.Any(), entity.BuildID{ID: testBuildID}).Return([]entity.BuildIssue{issue}, nil) - persisted := m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected) - m.expectNotification(t, true, persisted) + m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected) require.NoError(t, c.Process(context.Background(), m.delivery(t))) } @@ -507,7 +466,6 @@ func TestProcessRetriesErrorsAndReloadsStatus(t *testing.T) { m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil) m.request.TerminalBuildID = testBuildID - m.expectNotification(t, true, nil) require.NoError(t, c.Process(context.Background(), m.delivery(t))) } @@ -554,65 +512,6 @@ func TestCancelledAnalysisDoesNotHold(t *testing.T) { } } -func TestProcessRetriesPublicationAfterPersistence(t *testing.T) { - for _, status := range []entity.BuildStatus{entity.BuildStatusRunning, entity.BuildStatusFailed} { - t.Run(string(status), func(t *testing.T) { - c, m := newTestController(t) - issue := entity.BuildIssue{TaskID: "validation", UnitID: "unit"} - publishErr := errs.NewRetryableError(assert.AnError) - var failedMessageID string - - detected := m.expectPass(status, []entity.BuildIssue{issue}, nil) - persisted := m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue).Return(nil).After(detected) - m.publisher.EXPECT().Publish(gomock.Any(), "issue_track", gomock.Any()). - DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { - failedMessageID = msg.ID - return publishErr - }).After(persisted) - - require.ErrorIs(t, c.Process(context.Background(), m.delivery(t)), publishErr) - - detected = m.expectPass(status, []entity.BuildIssue{issue}, nil) - persisted = m.issues.EXPECT().Create(gomock.Any(), testBuildID, issue). - Return(storage.ErrAlreadyExists).After(detected) - published := m.expectNotification(t, status.IsTerminal(), persisted). - Do(func(_ context.Context, _ string, msg entityqueue.Message) { - if status.IsTerminal() { - assert.Equal(t, failedMessageID, msg.ID) - } else { - assert.NotEqual(t, failedMessageID, msg.ID) - } - }) - d := m.delivery(t) - if !status.IsTerminal() { - d.EXPECT().Hold(PollDelayMs).After(published) - } - - require.NoError(t, c.Process(context.Background(), d)) - }) - } -} - -func TestProcessPublishesDistinctRefreshesFromHeldMessage(t *testing.T) { - c, m := newTestController(t) - var messageIDs []string - - for range 2 { - detected := m.expectPass(entity.BuildStatusRunning, nil, nil) - published := m.expectNotification(t, false, detected). - Do(func(_ context.Context, _ string, msg entityqueue.Message) { - messageIDs = append(messageIDs, msg.ID) - }) - d := m.delivery(t) - d.EXPECT().Hold(PollDelayMs).After(published) - - require.NoError(t, c.Process(context.Background(), d)) - } - - require.Len(t, messageIDs, 2) - assert.NotEqual(t, messageIDs[0], messageIDs[1]) -} - func TestProcessWaitsForWinningBuildBeforeFinalizing(t *testing.T) { for _, outcome := range []string{"pending", "recorded during detection", "abandoned", "another build wins"} { t.Run(outcome, func(t *testing.T) { @@ -635,9 +534,8 @@ func TestProcessWaitsForWinningBuildBeforeFinalizing(t *testing.T) { d := m.delivery(t) if outcome != "another build wins" { - published := m.expectNotification(t, outcome == "recorded during detection", detected) if outcome == "pending" { - d.EXPECT().Hold(PollDelayMs).After(published) + d.EXPECT().Hold(PollDelayMs).After(detected) } } @@ -649,9 +547,8 @@ func TestProcessWaitsForWinningBuildBeforeFinalizing(t *testing.T) { require.Empty(t, m.history) } if outcome == "pending" { - next := m.expectPass(entity.BuildStatusFailed, nil, nil) + m.expectPass(entity.BuildStatusFailed, nil, nil) m.request.State = entity.RequestStateFailed - m.expectNotification(t, true, next) require.NoError(t, c.Process(context.Background(), m.delivery(t))) } diff --git a/stovepipe/controller/buildanalysis/history_test.go b/stovepipe/controller/buildanalysis/history_test.go index 2aa5b8763..0d8943d9c 100644 --- a/stovepipe/controller/buildanalysis/history_test.go +++ b/stovepipe/controller/buildanalysis/history_test.go @@ -22,7 +22,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" @@ -31,19 +30,17 @@ import ( func TestCompletedAnalysisHistoryRepairsOnRetry(t *testing.T) { for _, tt := range []struct { - name string - firstCount int - retryCount int - wantCount int - waitForWinner bool - failHistory bool - failPublication bool + name string + firstCount int + retryCount int + wantCount int + waitForWinner bool + failHistory bool }{ {name: "empty analysis", wantCount: 0}, {name: "idempotent replay", firstCount: 1, retryCount: 1, wantCount: 1}, {name: "winner recorded after held pass", firstCount: 1, retryCount: 2, wantCount: 2, waitForWinner: true}, {name: "history failure", firstCount: 1, retryCount: 2, wantCount: 2, failHistory: true}, - {name: "publication failure with more issues", firstCount: 1, retryCount: 2, wantCount: 1, failPublication: true}, {name: "replay with fewer issues", firstCount: 2, retryCount: 1, wantCount: 2}, } { t.Run(tt.name, func(t *testing.T) { @@ -73,7 +70,6 @@ func TestCompletedAnalysisHistoryRepairsOnRetry(t *testing.T) { } waiting := pass == 0 && tt.waitForWinner historyFailure := pass == 0 && tt.failHistory - publicationFailure := pass == 0 && tt.failPublication if waiting { m.request.TerminalBuildID = "" } @@ -81,23 +77,13 @@ func TestCompletedAnalysisHistoryRepairsOnRetry(t *testing.T) { *m.historyFailure = assert.AnError } else { *m.historyFailure = nil - last = m.expectNotification(t, !waiting, last).Do(func(context.Context, string, entityqueue.Message) { - if waiting { - require.Empty(t, m.history) - } else { - require.Len(t, m.history, 1) - } - }) - if publicationFailure { - last.Return(assert.AnError) - } } d := m.delivery(t) if waiting { d.EXPECT().Hold(PollDelayMs).After(last) } err := c.Process(context.Background(), d) - if historyFailure || publicationFailure { + if historyFailure { require.ErrorIs(t, err, assert.AnError) } else { require.NoError(t, err) @@ -151,8 +137,7 @@ func TestCompletedAnalysisHistoryRetainsConcurrentSnapshot(t *testing.T) { return tt.createErr }) m.historyCreate.After(created) - detected := m.expectPass(entity.BuildStatusFailed, nil, nil) - m.expectNotification(t, true, detected) + m.expectPass(entity.BuildStatusFailed, nil, nil) require.NoError(t, c.Process(context.Background(), m.delivery(t))) require.Len(t, m.history, 1) require.Equal(t, retained, m.history[retained.ID]) diff --git a/stovepipe/controller/issuetrack/README.md b/stovepipe/controller/issuetrack/README.md index 87e7bed73..a5bed0646 100644 --- a/stovepipe/controller/issuetrack/README.md +++ b/stovepipe/controller/issuetrack/README.md @@ -2,6 +2,18 @@ Issue tracking maintains current issue state and opening/closing history from persisted build issues. Each delivery applies a bounded batch of issues and requests immediate redelivery when more work remains. The reference service registers both the normal and recovery consumers. +## Validation order + +Admission registers one request-scoped tracking job before committing the request to processing or dispatching its build. A queue-version check after registration prevents a delayed publisher from admitting an older validation behind a newer registration. A request skipped after registration leaves a harmless job that acknowledges its terminal, unbuilt request. Build dispatch retains the request-to-build association before starting status polling and analysis; retries reuse that association. + +The tracking subscription must fetch only the first unacknowledged message in each queue partition. The reference host uses the shared `SingleMessageSubscriptionConfig`, whose message batch size is one; reconciliation still processes up to 100 issue keys per delivery. A larger message batch can prefetch later requests before the current job establishes its hold. Build execution and per-build detection remain concurrent. + +A request's job stays at the queue head through ordinary refreshes. Successful refreshes hold for 30 seconds; unfinished batches request immediate redelivery. The current request cannot be replaced until it is finalized or safely abandoned. A later request that reaches processing during recovery waits for that boundary rather than retrying the failed request's remaining work. + +Terminal readiness comes from the retained `build_analysis_completed` event, written only after the complete terminal issue set has been saved. `build_analysis_abandoned` explicitly ends incomplete analysis without allowing absence-based closure. These are point reads by the request and deterministic event ID, not scans of history. A completion event takes precedence if duplicate analysis deliveries retained both events. Neither CI status nor an empty issue set establishes readiness. The request, build, issue set, and analysis history must remain available until tracking finishes. + +This preserves a close in one validation followed by a new opening in the next, including its first-seen build and fixer eligibility. The next build's issues can accumulate in `build_issue` while its tracking job waits, but they do not change shared issue state or trigger repair until that job becomes current. + ## Applying one issue 1. Read whether the build reports the issue and load its current state. @@ -22,9 +34,9 @@ The post-read cursor write and processed-build updates, including unchanged stat ## Refreshing and finalizing -Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal notification retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh even when helping it through another notification. Encountering a task-wide issue (empty unit ID) sets `HasTaskWideIssues` in that same versioned record. This fact remains set across batches and retries, including when the issue was already applied by an earlier pass. +Every pass starts in `refreshing`, applying the build's accumulated issues in primary-key order. Reported issues are opened or kept open; nothing is closed. A terminal pass retains `Finalize = true` in the progress record, so a replacement worker can distinguish terminal work from an ordinary refresh. Encountering a task-wide issue (empty unit ID) sets `HasTaskWideIssues` in that same versioned record. This fact remains set across batches and retries, including when the issue was already applied by an earlier pass. -Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity combines the build and persisted pass notification IDs: retries share an identity, while later passes can trigger another evaluation. The fix controller selects an eligible issue and submits at most one repair per build. +Once the entire build-issue range is exhausted, an opening pass with a nonempty task cursor publishes a build-scoped repair request before saving its next phase. Empty passes and closing sweeps do not publish. Publication failures leave the pass active so retries repeat the handoff after ensuring issue state and history are durable. The repair message identity includes the persisted checkpoint version: retries before phase completion share an identity, while later refreshes of one long-lived job can trigger another evaluation without depending on queue garbage collection. The fix controller selects an eligible issue and submits at most one repair per build. Once the entire build-issue range is exhausted: @@ -40,7 +52,7 @@ No closure occurs before all accumulated build issues have been processed. A tas Every delivery processes at most 100 task/unit keys from one ordered range, including a revisited cursor and retained closed rows. Refreshing reads build issues; finalizing reads issue states. The cursor is always inclusive: revisit it to finish the issue and its history, then continue after it. A full batch retains its active phase and requests redelivery. Moving from refreshing to finalizing conditionally saves a reset cursor before any closing work begins. -An unfinished pass completes or is safely abandoned before another source is admitted. A notification that helps finish a different pass requests redelivery before handling its own source. Completed notifications and older sources are ignored, as are refreshes after the same Request has finalized or its terminal pass was abandoned. A terminal pass requires a terminal build chosen by its Request in both active phases. +A validation remains current after a completed refresh. Its job is acknowledged only after finalization or whole-validation abandonment; older jobs and duplicate completed jobs are ignored. A terminal pass requires a terminal build chosen by its Request in both active phases. Build IDs are opaque; source order comes from Request IDs. All scans are bounded reads under leading primary-key prefixes. There is no transaction spanning progress, issue state, and history. @@ -50,16 +62,16 @@ A finalized checkpoint retains `issues_reconciled` in its Request's history befo An error returns to normal consumer retry. The next delivery resumes the cursor before advancing. Once that retry policy dead-letters a notification, the recovery controller abandons the failed pass rather than rerunning its remaining batches: -1. Conditionally enter `abandoning`, retaining the source, notification, and cursor. Normal deliveries hold for 30 seconds while this phase is present; they cannot start another pass or resume issue application. +1. Conditionally enter `abandoning`, retaining the source, notification, and cursor. Later requests hold for 30 seconds while this phase is present. The owning request can resume abandonment cleanup, but cannot restart issue application. 2. If there is a cursor, read its issue and conditionally advance progress after that read. Advance the issue's version without changing its status or transition metadata. This invalidates a paused worker's earlier row guard. If the row has not been created yet, finish that one opening instead: the normal worker can only have saved such a cursor after finding a reported build issue. Creating it prevents the delayed create from succeeding. Closed issue rows are never deleted. 3. Repair the cursor's opening or closing history, summary, and open-issue lookup from the retained row. Record an `issue_tracking_abandoned` Request-history event keyed by the failed notification, with its build ID. 4. Conditionally enter `abandoned`. Keep the source-order boundary and notification identity, so the same delivery and older builds cannot restart the abandoned work. -Accepted opens and closes remain in place. Recovery does not scan either issue range, close the remaining issues, change Request or Build outcomes, or release build slots. An abandoned ordinary refresh still permits a new refresh or terminal notification for that Request; an abandoned terminal pass permits only newer Requests. A later successful build can reconcile issues left unchanged by the failed pass. +Accepted opens and closes remain in place. Recovery does not scan either issue range, close the remaining issues, change Request or Build outcomes, or release build slots. Abandoning a request-scoped job ends that validation's issue tracking, even if only an ordinary refresh had started. A pre-build abandonment retains the request boundary with an empty build ID and no issue cursor, fencing delayed workers before newer requests proceed. A later successful build can reconcile issues left unchanged by the failed pass. Recovery must finish fencing the current issue and retaining its history before admitting newer work. If a normal write wins the issue CAS first, recovery retries against that accepted row. If a replacement recovery worker advances first, its issue-version change rejects the old write. Merely marking the pass abandoned without touching the cursor row would not provide this protection. -A notification can fail while helping an older pass. Recovery abandons that saved pass first, then requests immediate redelivery to abandon the failed notification's own work if it has not already been superseded. A delayed DLQ delivery never abandons a newer build or a different notification from the same Request. +If a newer job reaches recovery while an earlier checkpoint is unfinished, recovery abandons that checkpoint first, then requests immediate redelivery to abandon the failed job's own work. A delayed DLQ delivery never abandons a newer request. Storage and history failures return for retry; hosts must pair recovery with an always-retryable, unlimited-retry DLQ consumer. Only recovery work repeats. Malformed message identities, unresolvable queues, and missing source records are logged and acknowledged. @@ -67,4 +79,10 @@ The reference service registers queue-partitioned `issuetrack` and `issuetrack_d ## Analysis handoff -Build analysis publishes refresh or final notifications after persisting build issues. Final notification requires successful terminal detection and the Request's recorded terminal build. Terminal detection returns the complete cumulative issue set, including issues returned by earlier calls, so finalization does not depend on earlier workers finishing their writes. Nonterminal detection may omit earlier issues without removing them from the build's accumulated results. The reference host uses a no-op detector; deployments supply their own detector and register the same normal and recovery paths. +Each validation uses the single job registered at admission; analysis does not publish tracking messages. Finalization requires successful terminal detection and the Request's recorded terminal build. Terminal detection returns the complete cumulative issue set, including issues returned by earlier calls, so finalization does not depend on earlier workers finishing their writes. Nonterminal detection may omit earlier issues without removing them from the build's accumulated results. The reference host uses a no-op detector; deployments supply their own detector and register the same normal and recovery paths. + +## Rollout + +Provision the non-null `request.build_id` column with an empty default before deploying. This is an additive association; existing terminal requests can still use their winning build ID. No new tables, secondary indexes, or cross-key transactions are required. + +Pause admission and drain pre-upgrade validations, completing or explicitly abandoning the existing tracking checkpoint before enabling request-scoped registration. Upgrade normal and recovery consumers together and use the single-head tracking subscription before resuming admission. Old per-analysis notifications are not supported. Do not mix old publishers or consumers with the request-scoped flow. To revert, pause admission and finish or safely abandon request-scoped jobs before restoring the old consumers; the added column can remain. diff --git a/stovepipe/controller/issuetrack/delivery_test.go b/stovepipe/controller/issuetrack/delivery_test.go index f3d66990a..3185f3ca9 100644 --- a/stovepipe/controller/issuetrack/delivery_test.go +++ b/stovepipe/controller/issuetrack/delivery_test.go @@ -24,16 +24,16 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" consumermock "github.com/uber/submitqueue/platform/consumer/mock" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/core/requestlog" "github.com/uber/submitqueue/stovepipe/entity" ) func (f *fixture) message(build string, final bool) entityqueue.Message { f.t.Helper() - f.sequence++ - payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{BuildId: build, QueueName: testQueue, Finalize: final}) - require.NoError(f.t, err) - msg := entityqueue.NewMessage(fmt.Sprintf("notice/%d", f.sequence), payload, testQueue, nil) - msg.Tenant = testQueue + msg := f.requestJob(build) + if final && f.builds[build].RequestID != "" { + f.analysisEvent(build, entity.RequestEventBuildAnalysisCompleted) + } return msg } @@ -43,6 +43,8 @@ func (f *fixture) deliver(msg entityqueue.Message, hold bool) error { delivery.EXPECT().Message().Return(msg) if hold { delivery.EXPECT().Hold(int64(0)) + } else { + delivery.EXPECT().Hold(_analysisWaitMs).MaxTimes(1) } return f.controller.Process(context.Background(), delivery) } @@ -62,8 +64,6 @@ func TestRefreshOpensReportedIssuesAndKeepsOtherState(t *testing.T) { f.reportIssue("build", "task", "next") require.NoError(t, f.deliver(msg, false)) - require.NotContains(t, f.states, issueKey{"task", "next"}) - require.NoError(t, f.deliver(f.message("build", false), false)) require.Contains(t, f.states, issueKey{"task", "next"}) require.Len(t, f.logs, 2) require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) @@ -211,19 +211,25 @@ func TestFinalNotificationBeforeLateRefreshAndOlderBuild(t *testing.T) { require.Empty(t, f.states) } -func TestAnotherNotificationFinishesTheCursorBeforeItsOwnBuild(t *testing.T) { +func TestNewerRequestWaitsForThePendingCursorAndFinalization(t *testing.T) { f := newFixture(t) f.source("build/100", "100", entity.BuildStatusRunning) f.source("build/101", "101", entity.BuildStatusSucceeded) f.reportIssue("build/100", "task", "unit") f.retain(currentCursor()) incoming := f.message("build/101", true) - require.NoError(t, f.deliver(incoming, true)) - require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) - require.Equal(t, "build/100", f.progress.BuildID) + require.NoError(t, f.deliverRequest(incoming, _analysisWaitMs)) + require.Equal(t, entity.IssueTrackingPhaseRefreshing, f.progress.Phase) + require.Empty(t, f.states) + older := f.message("build/100", false) + require.NoError(t, f.deliverRequest(older, _analysisWaitMs)) require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) - require.NoError(t, f.deliver(incoming, true)) - require.NoError(t, f.deliver(incoming, false)) + f.source("build/100", "100", entity.BuildStatusSucceeded) + f.analysisEvent("build/100", entity.RequestEventBuildAnalysisCompleted) + require.NoError(t, f.deliverRequest(older, 0)) + require.NoError(t, f.deliverRequest(older, -1)) + require.NoError(t, f.deliverRequest(incoming, 0)) + require.NoError(t, f.deliverRequest(incoming, -1)) require.Equal(t, "build/101", f.progress.BuildID) require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) require.Equal(t, int32(2), f.states[issueKey{"task", "unit"}].Version) @@ -308,29 +314,29 @@ func TestReadFailuresDoNotChangeIssues(t *testing.T) { } func TestFinalizationRequiresTheWinningTerminalBuild(t *testing.T) { - for _, status := range []entity.BuildStatus{entity.BuildStatusRunning, entity.BuildStatusSucceeded} { - t.Run(string(status), func(t *testing.T) { - f := newFixture(t) - f.source("build", "10", status) - request := f.requests["10"] - request.TerminalBuildID = "" - f.requests["10"] = request - require.Error(t, f.deliver(f.message("build", true), false)) - require.False(t, f.hasProgress) - require.Empty(t, f.states) - }) - } f := newFixture(t) - f.source("loser", "10", entity.BuildStatusFailed) - request := f.requests["10"] - request.TerminalBuildID = "winner" - f.requests["10"] = request - require.NoError(t, f.deliver(f.message("loser", true), false)) + f.source("build", "10", entity.BuildStatusRunning) + require.Error(t, f.deliver(f.message("build", true), false)) require.False(t, f.hasProgress) + require.Empty(t, f.states) +} + +func TestTrackingUsesTheRequestsWinningBuild(t *testing.T) { + f := newFixture(t) + f.source("loser", "10", entity.BuildStatusFailed) + job := f.message("loser", false) + f.reportIssue("loser", "ignored", "unit") + f.source("winner", "10", entity.BuildStatusSucceeded) + f.analysisEvent("winner", entity.RequestEventBuildAnalysisCompleted) + require.NoError(t, f.deliverRequest(job, 0)) + require.NoError(t, f.deliverRequest(job, -1)) + require.Equal(t, "winner", f.progress.BuildID) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) + require.Empty(t, f.states) } func TestProcessRejectsInvalidMessageAndSource(t *testing.T) { - for _, kind := range []string{"payload", "partition", "tenant", "missing message ID", "missing build", "request queue", "invalid request ID"} { + for _, kind := range []string{"payload", "partition", "tenant", "missing message ID", "missing queue", "missing source ID", "missing build", "request queue", "invalid request ID"} { t.Run(kind, func(t *testing.T) { f := newFixture(t) f.source("build", "10", entity.BuildStatusRunning) @@ -344,6 +350,12 @@ func TestProcessRejectsInvalidMessageAndSource(t *testing.T) { msg.Tenant = "other" case "missing message ID": msg.ID = "" + case "missing queue": + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{RequestId: "10"}) + require.NoError(t, err) + msg.Payload = payload + case "missing source ID": + msg = f.message("", false) case "missing build": delete(f.builds, "build") case "request queue": @@ -530,7 +542,7 @@ func TestTaskWideFlagSurvivesNoOpOpeningBatchAndRetry(t *testing.T) { } } -func TestAnotherNotificationRetainsTheSavedTerminalIntent(t *testing.T) { +func TestResumedPassRetainsTheSavedTerminalIntent(t *testing.T) { f := newFixture(t) f.source("build/100", "100", entity.BuildStatusFailed) f.source("build/101", "101", entity.BuildStatusRunning) @@ -541,13 +553,15 @@ func TestAnotherNotificationRetainsTheSavedTerminalIntent(t *testing.T) { progress.Finalize = true f.retain(progress) msg := f.message("build/101", false) + require.NoError(t, f.deliverRequest(msg, _analysisWaitMs)) + older := f.message("build/100", false) - require.NoError(t, f.deliver(msg, true)) + require.NoError(t, f.deliver(older, true)) require.Equal(t, "build/100", f.progress.BuildID) require.Equal(t, entity.IssueTrackingPhaseFinalizing, f.progress.Phase) require.True(t, f.progress.Finalize) require.Empty(t, f.progress.CursorTaskID) - require.NoError(t, f.deliver(msg, true)) + require.NoError(t, f.deliver(older, false)) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) require.NoError(t, f.deliver(msg, false)) require.Equal(t, "build/101", f.progress.BuildID) @@ -617,3 +631,208 @@ func TestRetryResetsCursorBeforeClosing(t *testing.T) { require.Equal(t, int32(1), f.states[issueKey{"task", "unit"}].Version) require.Len(t, f.logs, 2) } + +func (f *fixture) requestJob(buildID string) entityqueue.Message { + f.t.Helper() + requestID := f.builds[buildID].RequestID + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{RequestId: requestID, QueueName: testQueue}) + require.NoError(f.t, err) + msg := entityqueue.NewMessage(requestID+"/issues", payload, testQueue, nil) + msg.Tenant = testQueue + return msg +} + +func (f *fixture) analysisEvent(buildID string, event entity.RequestEvent) { + f.t.Helper() + request := f.requests[f.builds[buildID].RequestID] + log := requestlog.NewRequestEventLog(request, event, buildID, map[string]string{requestlog.MetadataKeyBuildID: buildID}) + log.TimestampMs = 1000 + f.requestHistory[request.ID+":"+log.ID] = log +} + +func (f *fixture) deliverRequest(msg entityqueue.Message, holdMs int64) error { + f.t.Helper() + delivery := consumermock.NewMockDelivery(f.mocks) + delivery.EXPECT().Message().Return(msg) + if holdMs >= 0 { + delivery.EXPECT().Hold(holdMs) + } + return f.controller.Process(context.Background(), delivery) +} + +func TestRequestJobWaitsForAnalysisThenPreservesReopening(t *testing.T) { + for _, newerTerminal := range []bool{false, true} { + t.Run(fmt.Sprint(newerTerminal), func(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusSucceeded) + status := entity.BuildStatusRunning + if newerTerminal { + status = entity.BuildStatusSucceeded + } + f.source("build/101", "101", status) + f.seedIssue("task", "unit", entity.IssueStatusOpen) + f.reportIssue("build/101", "task", "unit") + older, newer := f.requestJob("build/100"), f.requestJob("build/101") + require.NoError(t, f.deliverRequest(older, _analysisWaitMs)) + if newerTerminal { + f.analysisEvent("build/101", entity.RequestEventBuildAnalysisCompleted) + } + require.NoError(t, f.deliverRequest(newer, _analysisWaitMs)) + require.Equal(t, "100", f.progress.RequestID) + require.Equal(t, "older", f.states[issueKey{"task", "unit"}].FirstSeenBuildID) + f.analysisEvent("build/100", entity.RequestEventBuildAnalysisCompleted) + require.NoError(t, f.deliverRequest(older, 0)) + require.NoError(t, f.deliverRequest(older, -1)) + require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/100", entity.IssueStatusClosed}) + wait := _analysisWaitMs + if newerTerminal { + wait = 0 + } + require.NoError(t, f.deliverRequest(newer, wait)) + require.Equal(t, "build/101", f.states[issueKey{"task", "unit"}].FirstSeenBuildID) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) + require.Contains(t, f.logs, logKey{issueKey{"task", "unit"}, "build/101", entity.IssueStatusOpen}) + require.NoError(t, f.deliverRequest(older, -1)) + require.Equal(t, "101", f.progress.RequestID) + }) + } +} + +func TestRequestJobWaitsForBuildAssociation(t *testing.T) { + for _, state := range []entity.RequestState{entity.RequestStateAccepted, entity.RequestStateProcessing, entity.RequestStateSuperseded, entity.RequestStateFailed} { + t.Run(string(state), func(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + msg := f.requestJob("build/100") + request := f.requests["100"] + request.BuildID = "" + request.State = state + f.requests["100"] = request + wait := int64(-1) + if !state.IsTerminal() { + wait = _analysisWaitMs + } + require.NoError(t, f.deliverRequest(msg, wait)) + if state == entity.RequestStateFailed { + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + } else { + require.False(t, f.hasProgress) + } + require.Empty(t, f.states) + }) + } +} + +func TestRequestJobAppliesLaterRefreshesWithoutCompletingTheJob(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + job := f.requestJob("build/100") + for _, unit := range []string{"a", "b"} { + f.reportIssue("build/100", "task", unit) + require.NoError(t, f.deliverRequest(job, _analysisWaitMs)) + require.Equal(t, "build/100", f.states[issueKey{"task", unit}].FirstSeenBuildID) + require.Equal(t, entity.IssueTrackingPhaseRefreshed, f.progress.Phase) + } +} + +func TestTrackingSeesAnalysisCompletedAfterTheRequestRead(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + job := f.requestJob("build/100") + build := f.builds["build/100"] + build.Status = entity.BuildStatusSucceeded + f.builds[build.ID] = build + f.onHistoryRead = func() { + f.onHistoryRead = nil + request := f.requests["100"] + request.State = entity.RequestStateSucceeded + request.TerminalBuildID = build.ID + f.requests[request.ID] = request + f.analysisEvent(build.ID, entity.RequestEventBuildAnalysisCompleted) + } + + require.NoError(t, f.deliverRequest(job, 0)) + require.NoError(t, f.deliverRequest(job, -1)) + require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) +} + +func TestFailedRequestAbandonsTrackingWithoutReadingAnalysisHistory(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + job := f.requestJob("build/100") + request := f.requests["100"] + request.State = entity.RequestStateFailed + f.requests["100"] = request + f.failures["request-history/get"] = 1 + + require.NoError(t, f.deliverRequest(job, -1)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Equal(t, 1, f.failures["request-history/get"]) +} + +func TestAnalysisAbandonmentFinishesTheCursorBeforeTheNextRequest(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.source("build/101", "101", entity.BuildStatusRunning) + older, newer := f.requestJob("build/100"), f.requestJob("build/101") + f.reportIssue("build/100", "task", "unit") + progress := currentCursor() + progress.NotificationID = older.ID + f.retain(progress) + f.analysisEvent("build/100", entity.RequestEventBuildAnalysisAbandoned) + require.NoError(t, f.deliverRequest(newer, _analysisWaitMs)) + require.NoError(t, f.deliverRequest(older, -1)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Equal(t, "100", f.progress.RequestID) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"task", "unit"}].Status) + require.NoError(t, f.deliverRequest(newer, _analysisWaitMs)) + require.Equal(t, "101", f.progress.RequestID) +} + +func TestAbandonedRequestCannotRestartAfterALateBuildAssociation(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + job := f.requestJob("build/100") + request := f.requests["100"] + request.BuildID, request.State = "", entity.RequestStateFailed + f.requests["100"] = request + require.NoError(t, f.deliverRequest(job, -1)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Empty(t, f.progress.BuildID) + before := f.progress + f.source("build/100", "100", entity.BuildStatusSucceeded) + request = f.requests["100"] + request.BuildID = "build/100" + f.requests["100"] = request + f.reportIssue("build/100", "task", "unit") + f.analysisEvent("build/100", entity.RequestEventBuildAnalysisCompleted) + require.NoError(t, f.deliverRequest(job, -1)) + require.Equal(t, before, f.progress) + require.Empty(t, f.states) +} + +func TestAnalysisAbandonmentResumesItsCleanupAfterHistoryFailure(t *testing.T) { + for _, associated := range []bool{false, true} { + t.Run(fmt.Sprint(associated), func(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + job := f.requestJob("build/100") + if associated { + progress := currentCursor() + progress.NotificationID = job.ID + f.retain(progress) + f.analysisEvent("build/100", entity.RequestEventBuildAnalysisAbandoned) + } else { + request := f.requests["100"] + request.BuildID, request.State = "", entity.RequestStateFailed + f.requests["100"] = request + } + f.failures["request-history/create"] = 1 + require.ErrorIs(t, f.deliverRequest(job, -1), assert.AnError) + require.Equal(t, entity.IssueTrackingPhaseAbandoning, f.progress.Phase) + require.NoError(t, f.deliverRequest(job, -1)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + }) + } +} diff --git a/stovepipe/controller/issuetrack/dlq.go b/stovepipe/controller/issuetrack/dlq.go index ec6c47043..84585366b 100644 --- a/stovepipe/controller/issuetrack/dlq.go +++ b/stovepipe/controller/issuetrack/dlq.go @@ -39,14 +39,12 @@ const _recoveryDelayMs int64 = 30000 // DLQController abandons incomplete issue tracking while retaining accepted issue changes. type DLQController struct { - controller *Controller - materializer requestlog.Materializer + controller *Controller } func NewDLQController(logger *zap.SugaredLogger, scope tally.Scope, stores storage.Factory, materializer requestlog.Materializer, topicKey consumer.TopicKey, consumerGroup string) *DLQController { return &DLQController{ - controller: NewController(logger, scope, stores, materializer, consumer.TopicRegistry{}, topicKey, consumerGroup), - materializer: materializer, + controller: NewController(logger, scope, stores, materializer, consumer.TopicRegistry{}, topicKey, consumerGroup), } } @@ -64,8 +62,16 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) c.controller.logger.Errorw("discarding issue tracking dlq message with invalid queue identity", "message_id", msg.ID, "error", err) return nil } - if msg.ID == "" || queue == "" || notification.GetBuildId() == "" || msg.PartitionKey != queue { - c.controller.logger.Errorw("discarding issue tracking dlq message with invalid identity", "message_id", msg.ID, "queue", queue) + if msg.ID == "" { + c.controller.logger.Errorw("discarding issue tracking dlq message without a message ID", "queue", queue) + return nil + } + if notification.GetRequestId() == "" { + c.controller.logger.Errorw("discarding issue tracking dlq message without a request ID", "message_id", msg.ID, "queue", queue) + return nil + } + if msg.PartitionKey != queue { + c.controller.logger.Errorw("discarding issue tracking dlq message from the wrong partition", "message_id", msg.ID, "queue", queue, "partition", msg.PartitionKey) return nil } @@ -74,20 +80,12 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) c.controller.logger.Errorw("discarding issue tracking dlq message for unresolvable queue", "queue", queue, "message_id", msg.ID, "error", err) return nil } - build, err := loader.ByID(ctx, notification.GetBuildId(), store.GetBuildStore().Get, "build") - if err != nil { - if errors.Is(err, storage.ErrNotFound) { - c.controller.logger.Errorw("discarding issue tracking dlq message for missing build", "message_id", msg.ID, "error", err) - return nil - } - return err + request, err := loader.ByID(ctx, notification.GetRequestId(), store.GetRequestStore().Get, "tracked request") + if errors.Is(err, storage.ErrNotFound) { + c.controller.logger.Errorw("discarding issue tracking dlq message for missing source", "message_id", msg.ID, "error", err) + return nil } - request, err := loader.ByID(ctx, build.RequestID, store.GetRequestStore().Get, "request") if err != nil { - if errors.Is(err, storage.ErrNotFound) { - c.controller.logger.Errorw("discarding issue tracking dlq message for missing request", "message_id", msg.ID, "error", err) - return nil - } return err } if request.Queue != queue { @@ -99,7 +97,7 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) return nil } - progress, err := c.abandonPass(ctx, store, request, notification, msg.ID) + progress, err := c.controller.abandonPass(ctx, store, request, msg.ID) if err != nil { return err } @@ -120,7 +118,7 @@ func (c *DLQController) Process(ctx context.Context, delivery consumer.Delivery) return nil } -func (c *DLQController) abandonPass(ctx context.Context, store storage.Storage, request entity.Request, notification *stovepipemq.IssueTrack, notificationID string) (entity.IssueTrackingState, error) { +func (c *Controller) abandonPass(ctx context.Context, store storage.Storage, request entity.Request, notificationID string) (entity.IssueTrackingState, error) { tracking := store.GetIssueTrackingStore() progress, err := tracking.Get(ctx) if err != nil && !errors.Is(err, storage.ErrNotFound) { @@ -130,42 +128,34 @@ func (c *DLQController) abandonPass(ctx context.Context, store storage.Storage, if err := progress.Validate(); err != nil { return entity.IssueTrackingState{}, fmt.Errorf("invalid issue tracking recovery: %w", err) } - if err := persistIssuesReconciledLog(ctx, c.materializer, store, request.Queue, progress); err != nil { + if err := persistIssuesReconciledLog(ctx, c.requestMaterializer, store, request.Queue, progress); err != nil { return entity.IssueTrackingState{}, err } - if progress.NotificationID == notificationID { - switch progress.Phase { - case entity.IssueTrackingPhaseRefreshed, entity.IssueTrackingPhaseFinalized: + order, err := entity.CompareRequestID(request.ID, progress.RequestID) + if err != nil { + return entity.IssueTrackingState{}, err + } + if order < 0 { + return entity.IssueTrackingState{}, nil + } + switch progress.Phase { + case entity.IssueTrackingPhaseFinalized, entity.IssueTrackingPhaseAbandoned: + if order == 0 { return progress, nil - case entity.IssueTrackingPhaseAbandoned: - return progress, c.recordAbandonment(ctx, store, request.Queue, progress) - } - } else { - order, err := entity.CompareRequestID(request.ID, progress.RequestID) - if err != nil { - return entity.IssueTrackingState{}, err - } - if order < 0 || (order == 0 && progress.Finalize) { - return entity.IssueTrackingState{}, nil - } - switch progress.Phase { - case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing, entity.IssueTrackingPhaseAbandoning: - if order == 0 { - return entity.IssueTrackingState{}, nil - } - default: - progress = entity.IssueTrackingState{Version: progress.Version} } + progress = entity.IssueTrackingState{Version: progress.Version} } } if progress.NotificationID == "" { - if request.TerminalBuildID != "" && request.TerminalBuildID != notification.GetBuildId() { - return entity.IssueTrackingState{}, nil + progress.RequestID = request.ID + progress.BuildID = request.BuildID + if request.TerminalBuildID != "" { + progress.BuildID = request.TerminalBuildID } - progress.RequestID, progress.BuildID, progress.NotificationID = request.ID, notification.GetBuildId(), notificationID - progress.Finalize = notification.GetFinalize() + progress.NotificationID = notificationID } + progress.Finalize = true progress.Phase = entity.IssueTrackingPhaseAbandoning if progress.Version == 0 { progress.Version = 1 @@ -182,7 +172,7 @@ func (c *DLQController) abandonPass(ctx context.Context, store storage.Storage, return c.finishAbandonment(ctx, store, request.Queue, progress) } -func (c *DLQController) finishAbandonment(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) (entity.IssueTrackingState, error) { +func (c *Controller) finishAbandonment(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) (entity.IssueTrackingState, error) { if progress.CursorTaskID != "" { issues := store.GetIssueStateStore() current, err := issues.Get(ctx, progress.CursorTaskID, progress.CursorUnitID) @@ -204,7 +194,7 @@ func (c *DLQController) finishAbandonment(ctx context.Context, store storage.Sto TaskID: progress.CursorTaskID, UnitID: progress.CursorUnitID, Status: entity.IssueStatusOpen, FirstSeenBuildID: progress.BuildID, LastProcessedBuildID: progress.BuildID, LastChangedBuildID: progress.BuildID, - ChangedAtMs: c.controller.nowMs(), Version: 1, + ChangedAtMs: c.nowMs(), Version: 1, } err = issues.Create(ctx, current) } else { @@ -220,7 +210,7 @@ func (c *DLQController) finishAbandonment(ctx context.Context, store storage.Sto if err != nil { return entity.IssueTrackingState{}, fmt.Errorf("fence recovery cursor: %w", err) } - if err := c.controller.recordHistory(ctx, store, current); err != nil { + if err := c.recordHistory(ctx, store, current); err != nil { return entity.IssueTrackingState{}, err } } @@ -229,16 +219,20 @@ func (c *DLQController) finishAbandonment(ctx context.Context, store storage.Sto return entity.IssueTrackingState{}, err } progress.Phase = entity.IssueTrackingPhaseAbandoned - return saveProgress(ctx, store.GetIssueTrackingStore(), progress) + checkpoint, err := saveProgress(ctx, store.GetIssueTrackingStore(), progress) + if err == nil { + c.logger.Infow("abandoned issue tracking", "queue", queue, "request_id", checkpoint.RequestID, "build_id", checkpoint.BuildID, "notification_id", checkpoint.NotificationID) + } + return checkpoint, err } -func (c *DLQController) recordAbandonment(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) error { +func (c *Controller) recordAbandonment(ctx context.Context, store storage.Storage, queue string, progress entity.IssueTrackingState) error { request := entity.Request{ID: progress.RequestID, Queue: queue} // Notification IDs can exceed the request log's 255-character key. identity := sha256.Sum256([]byte(progress.NotificationID)) log := requestlog.NewRequestEventLog(request, entity.RequestEventIssueTrackingAbandoned, fmt.Sprintf("%x", identity), map[string]string{requestlog.MetadataKeyBuildID: progress.BuildID}) - if err := c.materializer.PersistLog(ctx, store, log); err != nil { + if err := c.requestMaterializer.PersistLog(ctx, store, log); err != nil { return fmt.Errorf("record incomplete issue tracking: %w", err) } return nil diff --git a/stovepipe/controller/issuetrack/dlq_test.go b/stovepipe/controller/issuetrack/dlq_test.go index 0c8ad3bd5..651eae630 100644 --- a/stovepipe/controller/issuetrack/dlq_test.go +++ b/stovepipe/controller/issuetrack/dlq_test.go @@ -27,6 +27,7 @@ import ( entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" "github.com/uber/submitqueue/platform/consumer" consumermock "github.com/uber/submitqueue/platform/consumer/mock" + stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" "github.com/uber/submitqueue/stovepipe/core/requestlog" requestlogmock "github.com/uber/submitqueue/stovepipe/core/requestlog/mock" "github.com/uber/submitqueue/stovepipe/entity" @@ -217,8 +218,10 @@ func TestRecoveryRetriesOnlyRecoveryWork(t *testing.T) { require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"task", "unit"}].Status) require.Len(t, f.events, 1) + f.failures["recovery/history"] = 1 require.NoError(t, f.recover(msg, false)) require.Len(t, f.events, 1) + require.Equal(t, 1, f.failures["recovery/history"]) }) } } @@ -307,24 +310,31 @@ func TestPrimaryWaitsForRecoveryThenAdmitsANewerBuild(t *testing.T) { require.Equal(t, "build/101", f.progress.BuildID) } -func TestRecoveryOfAHelperAbandonsTheSavedPassFirst(t *testing.T) { - f := newRecoveryFixture(t) - f.source("build/100", "100", entity.BuildStatusRunning) - f.source("build/101", "101", entity.BuildStatusFailed) - f.retain(currentCursor()) - msg := f.message("build/101", true) +func TestRecoveryAbandonsEarlierUnfinishedRequestFirst(t *testing.T) { + for _, phase := range []entity.IssueTrackingPhase{entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseRefreshed} { + t.Run(string(phase), func(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.source("build/101", "101", entity.BuildStatusFailed) + progress := currentCursor() + progress.Phase = phase + f.retain(progress) + msg := f.message("build/101", true) - require.NoError(t, f.recover(msg, true)) - require.Equal(t, "build/100", f.progress.BuildID) - require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) - require.NoError(t, f.recover(msg, false)) - require.Equal(t, "build/101", f.progress.BuildID) - require.True(t, f.progress.Finalize) - require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) - require.Len(t, f.events, 2) + require.NoError(t, f.recover(msg, true)) + require.Equal(t, "build/100", f.progress.BuildID) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Len(t, f.events, 1) + require.NoError(t, f.recover(msg, false)) + require.Equal(t, "build/101", f.progress.BuildID) + require.True(t, f.progress.Finalize) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Len(t, f.events, 2) + }) + } } -func TestTerminalNotificationCanFollowAnAbandonedRefresh(t *testing.T) { +func TestAbandonedRequestIgnoresLaterAnalysisCompletion(t *testing.T) { for _, terminal := range []bool{false, true} { t.Run(map[bool]string{false: "abandoned refresh", true: "abandoned terminal pass"}[terminal], func(t *testing.T) { f := newRecoveryFixture(t) @@ -338,16 +348,9 @@ func TestTerminalNotificationCanFollowAnAbandonedRefresh(t *testing.T) { require.NoError(t, f.recover(failed, false)) final := f.message("build/100", true) - if terminal { - require.NoError(t, f.deliver(final, false)) - require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) - require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"previous", "unit"}].Status) - } else { - require.NoError(t, f.deliver(final, true)) - require.NoError(t, f.deliver(final, false)) - require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - require.Equal(t, entity.IssueStatusClosed, f.states[issueKey{"previous", "unit"}].Status) - } + require.NoError(t, f.deliverRequest(final, -1)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"previous", "unit"}].Status) }) } } @@ -379,7 +382,7 @@ func TestRecoveryHistorySupportsLongNotificationIDs(t *testing.T) { } func TestRecoveryIgnoresCompletedAndUnrelatedPasses(t *testing.T) { - for _, scenario := range []string{"completed refresh", "completed terminal pass", "newer build", "different refresh"} { + for _, scenario := range []string{"completed terminal pass", "newer build"} { t.Run(scenario, func(t *testing.T) { f := newRecoveryFixture(t) f.source("build/100", "100", entity.BuildStatusFailed) @@ -387,14 +390,10 @@ func TestRecoveryIgnoresCompletedAndUnrelatedPasses(t *testing.T) { progress := currentCursor() progress.NotificationID = msg.ID switch scenario { - case "completed refresh": - progress.Phase = entity.IssueTrackingPhaseRefreshed case "completed terminal pass": progress.Phase, progress.Finalize = entity.IssueTrackingPhaseFinalized, true case "newer build": progress.RequestID, progress.BuildID, progress.NotificationID = "101", "build/101", "newer" - case "different refresh": - progress.NotificationID = "another-notification" } f.retain(progress) require.NoError(t, f.recover(msg, false)) @@ -406,7 +405,7 @@ func TestRecoveryIgnoresCompletedAndUnrelatedPasses(t *testing.T) { } func TestRecoveryAcknowledgesUnusableMessages(t *testing.T) { - for _, scenario := range []string{"payload", "queue", "partition", "message ID", "missing build", "missing request", "request queue", "request ID"} { + for _, scenario := range []string{"payload", "queue", "partition", "message ID", "missing queue", "missing source ID", "missing request", "request queue", "request ID"} { t.Run(scenario, func(t *testing.T) { f := newRecoveryFixture(t) f.source("build/100", "100", entity.BuildStatusFailed) @@ -420,8 +419,12 @@ func TestRecoveryAcknowledgesUnusableMessages(t *testing.T) { msg.PartitionKey = "another-queue" case "message ID": msg.ID = "" - case "missing build": - delete(f.builds, "build/100") + case "missing queue": + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{RequestId: "100"}) + require.NoError(t, err) + msg.Payload = payload + case "missing source ID": + msg = f.message("", true) case "missing request": delete(f.requests, "100") case "request queue": @@ -439,3 +442,33 @@ func TestRecoveryAcknowledgesUnusableMessages(t *testing.T) { }) } } + +func TestRecoveryAbandonsARequestBeforeBuildAssociation(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + job := f.requestJob("build/100") + request := f.requests["100"] + request.BuildID = "" + f.requests["100"] = request + require.NoError(t, f.recover(job, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.Equal(t, "100", f.progress.RequestID) + require.Empty(t, f.progress.BuildID) + require.Empty(t, f.states) + require.Len(t, f.events, 1) +} + +func TestRecoveryAbandonsAnEmptyRefreshWithoutItsBuildRecord(t *testing.T) { + f := newRecoveryFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + job := f.message("build/100", false) + progress := currentCursor() + progress.Phase = entity.IssueTrackingPhaseRefreshed + progress.CursorTaskID, progress.CursorUnitID = "", "" + f.retain(progress) + delete(f.builds, "build/100") + require.NoError(t, f.recover(job, false)) + require.Equal(t, entity.IssueTrackingPhaseAbandoned, f.progress.Phase) + require.True(t, f.progress.Finalize) + require.Len(t, f.events, 1) +} diff --git a/stovepipe/controller/issuetrack/fix_handoff_test.go b/stovepipe/controller/issuetrack/fix_handoff_test.go index 2239e0eae..2a3aecef9 100644 --- a/stovepipe/controller/issuetrack/fix_handoff_test.go +++ b/stovepipe/controller/issuetrack/fix_handoff_test.go @@ -25,7 +25,7 @@ import ( "github.com/uber/submitqueue/stovepipe/entity" ) -func TestOpeningPassPublishesOneBuildRepairRequest(t *testing.T) { +func TestOpeningPassPublishesRepairAfterIssueWrites(t *testing.T) { for _, tt := range []struct { name string status entity.BuildStatus @@ -64,7 +64,11 @@ func TestOpeningPassPublishesOneBuildRepairRequest(t *testing.T) { require.Len(t, f.fixMessages, 1) } require.NoError(t, f.deliver(msg, false)) - require.Len(t, f.fixMessages, 1) + if tt.final { + require.Len(t, f.fixMessages, 1) + } else { + require.Len(t, f.fixMessages, 2) + } }) } } @@ -159,7 +163,7 @@ func TestRepairHandoffRetriesBeforeCompletingPass(t *testing.T) { } } -func TestRepairHandoffUsesResumedBuildAndPassIdentity(t *testing.T) { +func TestRepairHandoffResumesBeforeTheNextRequest(t *testing.T) { f := newFixture(t) f.source("build", "10", entity.BuildStatusRunning) f.source("next-build", "11", entity.BuildStatusRunning) @@ -169,20 +173,24 @@ func TestRepairHandoffUsesResumedBuildAndPassIdentity(t *testing.T) { require.ErrorIs(t, f.deliver(msg, false), assert.AnError) nextMessage := f.message("next-build", false) - require.NoError(t, f.deliver(nextMessage, true)) + require.NoError(t, f.deliverRequest(nextMessage, _analysisWaitMs)) require.Equal(t, "build", f.progress.BuildID) + require.Len(t, f.fixMessages, 1) + require.NoError(t, f.deliverRequest(msg, _analysisWaitMs)) require.Len(t, f.fixMessages, 2) require.Equal(t, f.fixMessages[0].ID, f.fixMessages[1].ID) require.Equal(t, "build", f.fixMessages[1].PartitionKey) - require.NoError(t, f.deliver(msg, false)) - require.Len(t, f.fixMessages, 2) + f.source("build", "10", entity.BuildStatusFailed) + f.analysisEvent("build", entity.RequestEventBuildAnalysisCompleted) + require.NoError(t, f.deliverRequest(msg, 0)) + require.NoError(t, f.deliverRequest(msg, -1)) f.reportIssue("next-build", "new", "unit") require.NoError(t, f.deliver(nextMessage, false)) - require.Len(t, f.fixMessages, 3) - require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[2].ID) - require.Equal(t, "next-build", f.fixMessages[2].PartitionKey) + require.Len(t, f.fixMessages, 4) + require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[3].ID) + require.Equal(t, "next-build", f.fixMessages[3].PartitionKey) } func TestTaskWideFailureStillPublishesRepair(t *testing.T) { @@ -195,3 +203,22 @@ func TestTaskWideFailureStillPublishesRepair(t *testing.T) { require.Equal(t, entity.IssueStatusOpen, f.states[issueKey{"existing", "unit"}].Status) require.Len(t, f.fixMessages, 1) } + +func TestRequestJobPublishesDistinctRepairPassesAndStableRetries(t *testing.T) { + f := newFixture(t) + f.source("build/100", "100", entity.BuildStatusRunning) + f.seedIssue("old", "unit", entity.IssueStatusOpen) + f.reportIssue("build/100", "old", "unit") + job := f.requestJob("build/100") + require.NoError(t, f.deliverRequest(job, _analysisWaitMs)) + require.Len(t, f.fixMessages, 1) + f.reportIssue("build/100", "new", "unit") + f.failOnPhase = entity.IssueTrackingPhaseRefreshed + require.ErrorIs(t, f.deliverRequest(job, -1), assert.AnError) + require.Len(t, f.fixMessages, 2) + require.NotEqual(t, f.fixMessages[0].ID, f.fixMessages[1].ID) + require.NoError(t, f.deliverRequest(job, _analysisWaitMs)) + require.Len(t, f.fixMessages, 3) + require.Equal(t, f.fixMessages[1].ID, f.fixMessages[2].ID) + require.Equal(t, "build/100", f.states[issueKey{"new", "unit"}].FirstSeenBuildID) +} diff --git a/stovepipe/controller/issuetrack/fixture_test.go b/stovepipe/controller/issuetrack/fixture_test.go index fe1c60b4a..cae7e9a8d 100644 --- a/stovepipe/controller/issuetrack/fixture_test.go +++ b/stovepipe/controller/issuetrack/fixture_test.go @@ -62,10 +62,10 @@ type fixture struct { builds map[string]entity.Build requests map[string]entity.Request buildIssues map[string]map[issueKey]entity.BuildIssue - sequence int - fixMessages []entityqueue.Message - onFixPublish func(entityqueue.Message) + fixMessages []entityqueue.Message + onFixPublish func(entityqueue.Message) + onHistoryRead func() } func newFixture(t *testing.T) *fixture { @@ -370,6 +370,12 @@ func newFixture(t *testing.T) *fixture { return nil }).AnyTimes() requestHistory.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, requestID, logID string) (entity.RequestLog, error) { + if err := f.fail("request-history/get"); err != nil { + return entity.RequestLog{}, err + } + if f.onHistoryRead != nil { + f.onHistoryRead() + } log, exists := f.requestHistory[requestID+":"+logID] if !exists { return entity.RequestLog{}, storage.ErrNotFound @@ -390,7 +396,7 @@ func (f *fixture) fail(name string) error { } func currentCursor() entity.IssueTrackingState { - return entity.IssueTrackingState{RequestID: "100", BuildID: "build/100", NotificationID: "notice/100", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} + return entity.IssueTrackingState{RequestID: "100", BuildID: "build/100", NotificationID: "100/issues", CursorTaskID: "task", CursorUnitID: "unit", Phase: entity.IssueTrackingPhaseRefreshing, Version: 1} } func (f *fixture) retain(progress entity.IssueTrackingState) { @@ -410,7 +416,7 @@ func sortKeys(keys []issueKey) { func (f *fixture) source(buildID, requestID string, status entity.BuildStatus) { f.builds[buildID] = entity.Build{ID: buildID, RequestID: requestID, Status: status, Version: 1} - request := entity.Request{ID: requestID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 1} + request := entity.Request{ID: requestID, Queue: testQueue, BuildID: buildID, State: entity.RequestStateProcessing, Version: 1} if status.IsTerminal() { request.TerminalBuildID = buildID request.State = entity.RequestStateFailed diff --git a/stovepipe/controller/issuetrack/history.go b/stovepipe/controller/issuetrack/history.go index 66a5489b6..98e42d26a 100644 --- a/stovepipe/controller/issuetrack/history.go +++ b/stovepipe/controller/issuetrack/history.go @@ -16,6 +16,7 @@ package issuetrack import ( "context" + "errors" "fmt" "github.com/uber/submitqueue/stovepipe/core/requestlog" @@ -23,6 +24,21 @@ import ( "github.com/uber/submitqueue/stovepipe/extension/storage" ) +func terminalAnalysisEvent(ctx context.Context, store storage.Storage, request entity.Request, buildID string) (entity.RequestEvent, error) { + for _, event := range []entity.RequestEvent{entity.RequestEventBuildAnalysisCompleted, entity.RequestEventBuildAnalysisAbandoned} { + log := requestlog.NewRequestEventLog(request, event, buildID, nil) + _, err := store.GetRequestLogStore().Get(ctx, request.ID, log.ID) + if errors.Is(err, storage.ErrNotFound) { + continue + } + if err != nil { + return entity.RequestEventUnknown, fmt.Errorf("read terminal analysis readiness: %w", err) + } + return event, nil + } + return entity.RequestEventUnknown, nil +} + func persistIssuesReconciledLog(ctx context.Context, materializer requestlog.Materializer, store storage.Storage, queue string, progress entity.IssueTrackingState) error { if progress.Phase != entity.IssueTrackingPhaseFinalized { return nil diff --git a/stovepipe/controller/issuetrack/history_test.go b/stovepipe/controller/issuetrack/history_test.go index 21ba6ca58..13a3c6bcc 100644 --- a/stovepipe/controller/issuetrack/history_test.go +++ b/stovepipe/controller/issuetrack/history_test.go @@ -57,7 +57,7 @@ func TestReconciliationHistoryClosureModes(t *testing.T) { if mode == "reconciled" { require.NoError(t, f.deliver(msg, false)) } - require.Len(t, f.requestHistory, 1) + require.Len(t, f.requestHistory, 2) log := f.requestHistory["10:event/issues_reconciled/build"] require.Equal(t, entity.RequestEventIssuesReconciled, log.Event) require.Equal(t, mode, log.Metadata[requestlog.MetadataKeyClosureMode]) @@ -80,7 +80,7 @@ func TestCompletedReconciliationRepairsHistoryBeforeReplacingCheckpoint(t *testi f.failures["request-history/create"] = 1 require.ErrorIs(t, f.deliver(msg, false), assert.AnError) require.Equal(t, entity.IssueTrackingPhaseFinalized, f.progress.Phase) - require.Empty(t, f.requestHistory) + require.NotContains(t, f.requestHistory, "10:event/issues_reconciled/old") completed := f.progress f.source("new", "11", entity.BuildStatusRunning) newer := f.message("new", false) @@ -91,7 +91,7 @@ func TestCompletedReconciliationRepairsHistoryBeforeReplacingCheckpoint(t *testi require.NoError(t, f.deliver(newer, false)) require.Equal(t, "new", f.progress.BuildID) require.Contains(t, f.requestHistory, "10:event/issues_reconciled/old") - require.Len(t, f.requestHistory, 1) // A nonterminal refresh adds no event. + require.Len(t, f.requestHistory, 2) // Only the older validation has completion and reconciliation events. } func TestIssueTrackingDLQRepairsCompletedPassInsteadOfAbandoning(t *testing.T) { @@ -105,6 +105,6 @@ func TestIssueTrackingDLQRepairsCompletedPassInsteadOfAbandoning(t *testing.T) { checkpoint := f.progress require.NoError(t, f.recoverHistory(msg, failure.Failure{})) require.Equal(t, checkpoint, f.progress) - require.Len(t, f.requestHistory, 1) + require.Len(t, f.requestHistory, 2) require.Contains(t, f.requestHistory, "10:event/issues_reconciled/build") } diff --git a/stovepipe/controller/issuetrack/issuetrack.go b/stovepipe/controller/issuetrack/issuetrack.go index 8ea62d643..a8ce3bd3b 100644 --- a/stovepipe/controller/issuetrack/issuetrack.go +++ b/stovepipe/controller/issuetrack/issuetrack.go @@ -37,8 +37,11 @@ import ( "go.uber.org/zap" ) -// Each delivery processes at most this many issue keys, including a revisited cursor and closed records. -const _issueBatchSize = 100 +const ( + // Each delivery processes at most this many issue keys, including a revisited cursor and closed records. + _issueBatchSize = 100 + _analysisWaitMs int64 = 30000 +) type issueKey struct{ task, unit string } @@ -79,8 +82,11 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if err := entityqueue.ValidatePayloadQueue(msg, queue); err != nil { return fmt.Errorf("invalid message identity: %w", err) } - if msg.ID == "" || queue == "" || notification.GetBuildId() == "" { - return fmt.Errorf("issue tracking requires message, queue, and build identifiers") + if msg.ID == "" { + return fmt.Errorf("issue tracking message ID must not be empty") + } + if notification.GetRequestId() == "" { + return fmt.Errorf("issue tracking requires a request ID") } if msg.PartitionKey != queue { return fmt.Errorf("issue tracking must be partitioned by queue %q", queue) @@ -89,74 +95,125 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er if err != nil { return fmt.Errorf("resolve issue tracking storage: %w", err) } - return c.trackBuild(ctx, delivery, store, notification, msg.ID) + return c.trackRequest(ctx, delivery, store, notification, msg.ID) } -func (c *Controller) trackBuild(ctx context.Context, delivery consumer.Delivery, store storage.Storage, notification *stovepipemq.IssueTrack, notificationID string) error { +func (c *Controller) trackRequest(ctx context.Context, delivery consumer.Delivery, store storage.Storage, notification *stovepipemq.IssueTrack, notificationID string) error { queue := notification.GetQueueName() - build, request, err := loadBuildAndRequest(ctx, store, queue, notification.GetBuildId()) - if err != nil { - return err - } - tracking := store.GetIssueTrackingStore() for { if err := ctx.Err(); err != nil { return err } + request, err := loader.ByID(ctx, notification.GetRequestId(), store.GetRequestStore().Get, "tracked request") + if err != nil { + return err + } + if request.Queue != queue { + return fmt.Errorf("tracked request belongs to another queue") + } + if err := id.Validate(request.ID); err != nil { + return err + } progress, err := tracking.Get(ctx) if errors.Is(err, storage.ErrNotFound) { progress = entity.IssueTrackingState{} } else if err != nil { return fmt.Errorf("load issue tracking progress: %w", err) } - // A completed checkpoint is the repair source until its history exists. - // Retain that occurrence before another notification can replace it. + // Finish the previous validation's completion history before replacing its checkpoint. if err := persistIssuesReconciledLog(ctx, c.requestMaterializer, store, queue, progress); err != nil { return err } - - if progress.Phase == entity.IssueTrackingPhaseAbandoning { - delivery.Hold(_recoveryDelayMs) - return nil - } - - if progress.Phase != entity.IssueTrackingPhaseRefreshing && progress.Phase != entity.IssueTrackingPhaseFinalizing { - if request.TerminalBuildID != "" && request.TerminalBuildID != build.ID { + if progress.Version != 0 { + order, err := entity.CompareRequestID(request.ID, progress.RequestID) + if err != nil { + return err + } + if order < 0 { return nil } - if progress.Version != 0 { - if progress.NotificationID == notificationID { - return nil - } - order, err := entity.CompareRequestID(request.ID, progress.RequestID) - if err != nil { + switch progress.Phase { + case entity.IssueTrackingPhaseAbandoning: + if order == 0 { + _, err := c.finishAbandonment(ctx, store, queue, progress) return err } - if order < 0 || (order == 0 && progress.Finalize) { - c.logger.Debugw("ignoring superseded issue tracking notification", - "queue", queue, "build_id", build.ID, "request_id", request.ID, "current_request_id", progress.RequestID) + delivery.Hold(_recoveryDelayMs) + return nil + case entity.IssueTrackingPhaseFinalized, entity.IssueTrackingPhaseAbandoned: + if order == 0 { + return nil + } + default: + if order > 0 { + delivery.Hold(_analysisWaitMs) return nil } } + } + buildID := request.BuildID + if request.TerminalBuildID != "" { + buildID = request.TerminalBuildID + } + if buildID == "" { + if request.State == entity.RequestStateSuperseded { + return nil + } + if !request.State.IsTerminal() { + delivery.Hold(_analysisWaitMs) + return nil + } + } + if request.State.IsTerminal() && request.TerminalBuildID == "" { + _, err := c.abandonPass(ctx, store, request, notificationID) + return err + } + event, err := terminalAnalysisEvent(ctx, store, request, buildID) + if err != nil { + return err + } + if event == entity.RequestEventBuildAnalysisAbandoned { + _, err := c.abandonPass(ctx, store, request, notificationID) + return err + } + finalize := event == entity.RequestEventBuildAnalysisCompleted + // Completion may have been recorded after the request read above. + if finalize && request.TerminalBuildID != buildID { + request, err = loader.ByID(ctx, request.ID, store.GetRequestStore().Get, "completed request") + if err != nil { + return err + } + } + build, err := loader.ByID(ctx, buildID, store.GetBuildStore().Get, "tracked build") + if err != nil { + return err + } + if build.RequestID != request.ID { + return fmt.Errorf("associated build belongs to another tracking request") + } + + switch progress.Phase { + case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing: + // Resume the saved pass before starting a newer refresh or finalization. + default: if progress.Version == math.MaxInt32 { return fmt.Errorf("issue tracking version exhausted") } + if finalize { + if err := requireTerminalBuild(build, request); err != nil { + return err + } + } next := entity.IssueTrackingState{ RequestID: request.ID, BuildID: build.ID, NotificationID: notificationID, Phase: entity.IssueTrackingPhaseRefreshing, - Finalize: notification.GetFinalize(), + Finalize: finalize, Version: progress.Version + 1, } - if next.Finalize { - if err := requireTerminalBuild(build, request); err != nil { - return err - } - } - if progress.Version == 0 { err = tracking.Create(ctx, next) } else { @@ -168,9 +225,14 @@ func (c *Controller) trackBuild(ctx context.Context, delivery consumer.Delivery, if err != nil { return fmt.Errorf("start issue tracking pass: %w", err) } + if progress.RequestID != request.ID { + c.logger.Infow("tracking validation", "queue", queue, "request_id", request.ID, "build_id", build.ID, "notification_id", notificationID) + } progress = next + if finalize { + c.logger.Infow("starting issue finalization", "queue", queue, "request_id", request.ID, "build_id", build.ID) + } } - checkpoint, err := c.reconcileBatch(ctx, store, queue, progress) if err != nil { return err @@ -178,10 +240,11 @@ func (c *Controller) trackBuild(ctx context.Context, delivery consumer.Delivery, if err := persistIssuesReconciledLog(ctx, c.requestMaterializer, store, queue, checkpoint); err != nil { return err } - if checkpoint.NotificationID != notificationID || - checkpoint.Phase == entity.IssueTrackingPhaseRefreshing || - checkpoint.Phase == entity.IssueTrackingPhaseFinalizing { + switch checkpoint.Phase { + case entity.IssueTrackingPhaseRefreshing, entity.IssueTrackingPhaseFinalizing: delivery.Hold(0) + case entity.IssueTrackingPhaseRefreshed: + delivery.Hold(_analysisWaitMs) } return nil } @@ -269,7 +332,9 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, if len(keys) < limit { if !finalizing && progress.CursorTaskID != "" { - if err := fixrequest.Publish(ctx, c.registry, queue, progress.BuildID, progress.NotificationID); err != nil { + // The checkpoint distinguishes refreshes of one long-lived tracking job. + repairID := fmt.Sprintf("%s/%d", progress.NotificationID, progress.Version) + if err := fixrequest.Publish(ctx, c.registry, queue, progress.BuildID, repairID); err != nil { return entity.IssueTrackingState{}, fmt.Errorf("publish build repair request: %w", err) } } @@ -285,7 +350,12 @@ func (c *Controller) reconcileBatch(ctx context.Context, store storage.Storage, next.Phase = entity.IssueTrackingPhaseFinalizing next.CursorTaskID, next.CursorUnitID = "", "" } - return saveProgress(ctx, store.GetIssueTrackingStore(), next) + checkpoint, err := saveProgress(ctx, store.GetIssueTrackingStore(), next) + if err == nil && checkpoint.Phase == entity.IssueTrackingPhaseFinalized { + c.logger.Infow("completed issue finalization", "queue", queue, "request_id", checkpoint.RequestID, "build_id", checkpoint.BuildID, + "has_task_wide_issues", checkpoint.HasTaskWideIssues, "status", build.Status) + } + return checkpoint, err } if cursorNeedsCheckpoint { return saveProgress(ctx, store.GetIssueTrackingStore(), progress) diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 6317bc646..10edcf83c 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -14,14 +14,15 @@ // Package process holds the process-stage queue controller. It consumes request // ids from ingest, reloads the Request from storage, coalesces older heads, and -// admits the latest head when a build slot is open. Build queue publish lands in -// a follow-up PR. +// admits the latest head when a build slot is open, registers issue tracking, +// and dispatches the build. package process import ( "context" "errors" "fmt" + "math" "github.com/uber-go/tally" basehook "github.com/uber/submitqueue/api/base/hook" @@ -237,6 +238,24 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, return err } + if err := c.publishIssueTracking(ctx, request); err != nil { + return err + } + // A delayed registration must not admit an older request behind a newer job. + if queueRow.Version == math.MaxInt32 { + return fmt.Errorf("queue version exhausted") + } + newVersion := queueRow.Version + 1 + err = store.GetQueueStore().Update(ctx, queueRow, queueRow.Version, newVersion) + if errors.Is(err, storage.ErrVersionMismatch) { + delivery.Hold(0) + return nil + } + if err != nil { + return fmt.Errorf("confirm tracking registration before admission: %w", err) + } + queueRow.Version = newVersion + // The request transition is the admit and the count follows it, so a failure between the two // writes leaves the count low (over-admission by one) rather than a slot nobody owns (a wedge). The // gate is therefore best-effort: two heads that read it open in the same window both pass, and @@ -461,6 +480,20 @@ func (c *Controller) publishBuild(ctx context.Context, id, queue string) error { return nil } +func (c *Controller) publishIssueTracking(ctx context.Context, request entity.Request) error { + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{RequestId: request.ID, QueueName: request.Queue}) + if err != nil { + return fmt.Errorf("serialize issue tracking registration: %w", err) + } + if err := publish.Message(ctx, c.registry, stovepipemq.TopicKeyIssueTrack, publish.MessageParams{ + Tenant: request.Queue, ID: publish.IntentID(request.ID, "issues"), + Payload: payload, PartitionKey: request.Queue, + }); err != nil { + return fmt.Errorf("register issue tracking for request %s: %w", request.ID, err) + } + return nil +} + // publishHookEvent announces a lifecycle transition on the hook topic. // // Published only once the transition is durable: the payload names the request diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index 2f5e87f9a..fa87d2668 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -56,13 +56,15 @@ func queueContext(queueName string) context.Context { } type processMocks struct { - reqStore *storagemock.MockRequestStore - queueStore *storagemock.MockQueueStore - sourceFactory *sourcecontrolmock.MockFactory - sourceControl *sourcecontrolmock.MockSourceControl - store *storagemock.MockStorage - materializer *requestlogmock.MockMaterializer - publisher *mqmock.MockPublisher + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore + sourceFactory *sourcecontrolmock.MockFactory + sourceControl *sourcecontrolmock.MockSourceControl + store *storagemock.MockStorage + materializer *requestlogmock.MockMaterializer + publisher *mqmock.MockPublisher + registration *gomock.Call + admissionGuard *gomock.Call } // staticStorageFactory resolves every queue to one fixed store aggregate. @@ -94,6 +96,7 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S queue := mqmock.NewMockQueue(ctrl) queue.EXPECT().Publisher().Return(m.publisher).AnyTimes() registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{ + {Key: stovepipemq.TopicKeyIssueTrack, Name: "issuetrack", Queue: queue}, {Key: stovepipemq.TopicKeyProcess, Name: "process", Queue: queue}, {Key: stovepipemq.TopicKeyBuild, Name: "build", Queue: queue}, {Key: basehook.TopicKeyHook, Name: "stovepipe-hook", Queue: queue}, @@ -111,6 +114,16 @@ func newControllerWithScope(t *testing.T, ctrl *gomock.Controller, scope tally.S stovepipemq.TopicKeyProcess, "stovepipe-process", ) + m.registration = m.publisher.EXPECT().Publish(gomock.Any(), "issuetrack", gomock.Any()).DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error { + var job stovepipemq.IssueTrack + require.NoError(t, stovepipemq.Unmarshal(msg.Payload, &job)) + require.Equal(t, testQueue, msg.PartitionKey) + require.Equal(t, job.RequestId+"/issues", msg.ID) + return nil + }).AnyTimes() + m.admissionGuard = m.queueStore.EXPECT().Update(gomock.Any(), gomock.Cond(func(q entity.Queue) bool { + return q.InFlightCount == 0 && q.Version == 1 + }), int32(1), int32(2)).Return(nil).AnyTimes() return c, m } @@ -186,8 +199,8 @@ func expectAdmitWrites(t *testing.T, m processMocks, id string) (entity.Request, admitted, requestCall := expectRequestAdmit(t, m, id) slotCall := m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: id, InFlightCount: 1, Version: 1, - }, int32(1), int32(2)).Return(nil).After(requestCall) + Name: testQueue, LatestRequestID: id, InFlightCount: 1, Version: 2, + }, int32(2), int32(3)).Return(nil).After(requestCall) return admitted, slotCall } @@ -503,6 +516,7 @@ func TestProcess(t *testing.T) { { name: "processing redelivery re-announces the start and republishes to build", setup: func(m processMocks) { + m.registration.Times(0) request := entity.Request{ ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, } @@ -514,6 +528,7 @@ func TestProcess(t *testing.T) { name: "processing redelivery stops before handoff when log persistence fails", wantErr: true, setup: func(m processMocks) { + m.registration.Times(0) request := entity.Request{ ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, } @@ -650,15 +665,15 @@ func TestProcess(t *testing.T) { }, nil) admitted, requestCall := expectRequestAdmit(t, m, testID) firstCount := m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, - }, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 2, + }, int32(2), int32(3)).Return(storage.ErrVersionMismatch) // A closed gate and a newer head on the reload change nothing: the admit is already durable. reload := m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, LatestRequestID: "9", InFlightCount: 1, Version: 2, + Name: testQueue, LatestRequestID: "9", InFlightCount: 1, Version: 3, }, nil) retryCount := m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: "9", InFlightCount: 2, Version: 2, - }, int32(2), int32(3)).Return(nil) + Name: testQueue, LatestRequestID: "9", InFlightCount: 2, Version: 3, + }, int32(3), int32(4)).Return(nil) gomock.InOrder(requestCall, firstCount, reload, retryCount) expectProcessingLogAndHandoff(t, m, admitted).After(retryCount) }, @@ -673,8 +688,8 @@ func TestProcess(t *testing.T) { }, nil) _, requestCall := expectRequestAdmit(t, m, testID) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, - }, int32(1), int32(2)).Return(errors.New("db down")).After(requestCall) + Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 2, + }, int32(2), int32(3)).Return(errors.New("db down")).After(requestCall) }, }, { @@ -705,8 +720,8 @@ func TestProcess(t *testing.T) { retry.BaseURI = lastGreenURI retryCall := m.reqStore.EXPECT().Update(gomock.Any(), retry, int32(2), int32(3)).Return(nil) slotCall := m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, LastGreenURI: lastGreenURI, Version: 1, - }, int32(1), int32(2)).Return(nil) + Name: testQueue, LatestRequestID: testID, InFlightCount: 1, LastGreenURI: lastGreenURI, Version: 2, + }, int32(2), int32(3)).Return(nil) gomock.InOrder(firstCall, reloadCall, retryCall, slotCall) loggedRequest := retry loggedRequest.Version = 3 @@ -901,3 +916,73 @@ func TestHoldForBuildSlotRequiresPositiveDelay(t *testing.T) { require.Error(t, err) assert.False(t, errs.IsRetryable(err)) } + +func TestTrackingRegistrationFailureDoesNotAdmitTheRequest(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, LatestRequestID: testID, Version: 1}, nil) + m.registration.Return(assert.AnError).Times(1) + require.ErrorIs(t, c.Process(queueContext(testQueue), delivery(t, ctrl, processPayload(t, testID))), assert.AnError) +} + +func TestDelayedRegistrationDoesNotAdmitAnOlderRequest(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil).Times(2) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, LatestRequestID: testID, Version: 1}, nil) + m.registration.Times(1) + m.admissionGuard.Return(storage.ErrVersionMismatch).Times(1) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, LatestRequestID: "8", Version: 2}, nil) + request := acceptedRequest(testID) + request.State = entity.RequestStateSuperseded + m.reqStore.EXPECT().Update(gomock.Any(), request, int32(1), int32(2)).Return(nil) + request.Version = 2 + expectStateLog(t, m, request, entity.RequestOutcomeReasonSupersededByNewerHead) + + first := delivery(t, ctrl, processPayload(t, testID)) + first.EXPECT().Hold(int64(0)) + require.NoError(t, c.Process(queueContext(testQueue), first)) + require.NoError(t, c.Process(queueContext(testQueue), delivery(t, ctrl, processPayload(t, testID)))) +} + +func TestRegistrationConflictRetriesAdmissionOnRedelivery(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil).Times(2) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, LatestRequestID: testID, Version: 1}, nil) + m.registration.Times(2) + m.admissionGuard.Return(storage.ErrVersionMismatch).Times(1) + + first := delivery(t, ctrl, processPayload(t, testID)) + first.EXPECT().Hold(int64(0)) + require.NoError(t, c.Process(queueContext(testQueue), first)) + + queue := entity.Queue{Name: testQueue, LatestRequestID: testID, Version: 2} + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queue, nil) + guard := m.queueStore.EXPECT().Update(gomock.Any(), queue, int32(2), int32(3)).Return(nil) + admitted, requestCAS := expectRequestAdmit(t, m, testID) + requestCAS.After(guard) + queue.Version = 3 + queue.InFlightCount = 1 + slotCAS := m.queueStore.EXPECT().Update(gomock.Any(), queue, int32(3), int32(4)).Return(nil).After(requestCAS) + expectProcessingLogAndHandoff(t, m, admitted).After(slotCAS) + + require.NoError(t, c.Process(queueContext(testQueue), delivery(t, ctrl, processPayload(t, testID)))) +} + +func TestTrackingRegistrationPrecedesAdmissionAndBuildDispatch(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, LatestRequestID: testID, Version: 1}, nil) + m.registration.Times(1) + m.admissionGuard.Times(1).After(m.registration) + admitted, requestCAS := expectRequestAdmit(t, m, testID) + requestCAS.After(m.admissionGuard) + slotCAS := m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 2, + }, int32(2), int32(3)).Return(nil).After(requestCAS) + expectProcessingLogAndHandoff(t, m, admitted).After(slotCAS) + require.NoError(t, c.Process(queueContext(testQueue), delivery(t, ctrl, processPayload(t, testID)))) +} diff --git a/stovepipe/core/messagequeue/messagequeue_test.go b/stovepipe/core/messagequeue/messagequeue_test.go index fd3e20ac2..c5b5af291 100644 --- a/stovepipe/core/messagequeue/messagequeue_test.go +++ b/stovepipe/core/messagequeue/messagequeue_test.go @@ -121,31 +121,19 @@ func TestBuildAnalysisRoundTrip(t *testing.T) { } func TestIssueTrackRoundTrip(t *testing.T) { - for _, tt := range []struct { - name string - finalize bool - wire string - }{ - {name: "observations", wire: `{"build_id":"build/42","queue_name":"validation/main"}`}, - {name: "final analysis", finalize: true, wire: `{"build_id":"build/42","queue_name":"validation/main","finalize":true}`}, - } { - t.Run(tt.name, func(t *testing.T) { - notification := &IssueTrack{BuildId: "build/42", QueueName: "validation/main", Finalize: tt.finalize} - data, err := Marshal(notification) - require.NoError(t, err) - assert.JSONEq(t, tt.wire, string(data)) - - got := &IssueTrack{} - require.NoError(t, Unmarshal(data, got)) - assert.True(t, proto.Equal(notification, got)) - }) - } + job := &IssueTrack{RequestId: "42", QueueName: "validation/main"} + data, err := Marshal(job) + require.NoError(t, err) + assert.JSONEq(t, `{"request_id":"42","queue_name":"validation/main"}`, string(data)) + got := &IssueTrack{} + require.NoError(t, Unmarshal(data, got)) + assert.True(t, proto.Equal(job, got)) } func TestIssueTrackToleratesUnknownFields(t *testing.T) { got := &IssueTrack{} - require.NoError(t, Unmarshal([]byte(`{"build_id":"build/42","queue_name":"validation/main","finalize":true,"future_field":42}`), got)) - assert.True(t, proto.Equal(&IssueTrack{BuildId: "build/42", QueueName: "validation/main", Finalize: true}, got)) + require.NoError(t, Unmarshal([]byte(`{"request_id":"42","queue_name":"validation/main","future_field":42}`), got)) + assert.True(t, proto.Equal(&IssueTrack{RequestId: "42", QueueName: "validation/main"}, got)) } func TestFixRequestRoundTrip(t *testing.T) { diff --git a/stovepipe/core/messagequeue/proto/issuetrack.proto b/stovepipe/core/messagequeue/proto/issuetrack.proto index 52acce9f5..2e32ef465 100644 --- a/stovepipe/core/messagequeue/proto/issuetrack.proto +++ b/stovepipe/core/messagequeue/proto/issuetrack.proto @@ -23,17 +23,16 @@ option java_multiple_files = true; option java_outer_classname = "IssueTrackProto"; option java_package = "com.uber.submitqueue.stovepipe.messagequeue"; -// IssueTrack requests queue-scoped reconciliation of persisted build observations. -// Publish only after successful detection and observation persistence; partition by queue. +// IssueTrack tracks one validation through its final issue reconciliation. +// Register in admission order before build dispatch; partition by queue. message IssueTrack { option (uber.base.messagequeue.topic_keys) = "issuetrack"; - // build_id is the required opaque identifier of the persisted build. - string build_id = 1; - // queue_name is the required queue namespace of the build. + reserved 1, 3; + reserved "build_id", "finalize"; + + // queue_name identifies the validation queue. string queue_name = 2; - // finalize requests final reconciliation after successful terminal detection - // and persistence, including when no failures were detected. - // Otherwise the notification permits opening issues, not resolution by absence. - bool finalize = 3; + // request_id identifies a validation tracked until finalization or abandonment. + string request_id = 4; } diff --git a/stovepipe/core/messagequeue/protopb/issuetrack.pb.go b/stovepipe/core/messagequeue/protopb/issuetrack.pb.go index 38a282c26..abefd0818 100644 --- a/stovepipe/core/messagequeue/protopb/issuetrack.pb.go +++ b/stovepipe/core/messagequeue/protopb/issuetrack.pb.go @@ -37,18 +37,14 @@ const ( _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) ) -// IssueTrack requests queue-scoped reconciliation of persisted build observations. -// Publish only after successful detection and observation persistence; partition by queue. +// IssueTrack tracks one validation through its final issue reconciliation. +// Register in admission order before build dispatch; partition by queue. type IssueTrack struct { state protoimpl.MessageState `protogen:"open.v1"` - // build_id is the required opaque identifier of the persisted build. - BuildId string `protobuf:"bytes,1,opt,name=build_id,json=buildId,proto3" json:"build_id,omitempty"` - // queue_name is the required queue namespace of the build. + // queue_name identifies the validation queue. QueueName string `protobuf:"bytes,2,opt,name=queue_name,json=queueName,proto3" json:"queue_name,omitempty"` - // finalize requests final reconciliation after successful terminal detection - // and persistence, including when no failures were detected. - // Otherwise the notification permits opening issues, not resolution by absence. - Finalize bool `protobuf:"varint,3,opt,name=finalize,proto3" json:"finalize,omitempty"` + // request_id identifies a validation tracked until finalization or abandonment. + RequestId string `protobuf:"bytes,4,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -83,13 +79,6 @@ func (*IssueTrack) Descriptor() ([]byte, []int) { return file_issuetrack_proto_rawDescGZIP(), []int{0} } -func (x *IssueTrack) GetBuildId() string { - if x != nil { - return x.BuildId - } - return "" -} - func (x *IssueTrack) GetQueueName() string { if x != nil { return x.QueueName @@ -97,25 +86,25 @@ func (x *IssueTrack) GetQueueName() string { return "" } -func (x *IssueTrack) GetFinalize() bool { +func (x *IssueTrack) GetRequestId() string { if x != nil { - return x.Finalize + return x.RequestId } - return false + return "" } var File_issuetrack_proto protoreflect.FileDescriptor const file_issuetrack_proto_rawDesc = "" + "\n" + - "\x10issuetrack.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"r\n" + + "\x10issuetrack.proto\x12\x1buber.stovepipe.messagequeue\x1a.api/base/messagequeue/proto/messagequeue.proto\"z\n" + + "\n" + + "IssueTrack\x12\x1d\n" + "\n" + - "IssueTrack\x12\x19\n" + - "\bbuild_id\x18\x01 \x01(\tR\abuildId\x12\x1d\n" + + "queue_name\x18\x02 \x01(\tR\tqueueName\x12\x1d\n" + "\n" + - "queue_name\x18\x02 \x01(\tR\tqueueName\x12\x1a\n" + - "\bfinalize\x18\x03 \x01(\bR\bfinalize:\x0e\x8a\xb5\x18\n" + - "issuetrackB\x81\x01\n" + + "request_id\x18\x04 \x01(\tR\trequestId:\x0e\x8a\xb5\x18\n" + + "issuetrackJ\x04\b\x01\x10\x02J\x04\b\x03\x10\x04R\bbuild_idR\bfinalizeB\x81\x01\n" + "+com.uber.submitqueue.stovepipe.messagequeueB\x0fIssueTrackProtoP\x01Z?github.com/uber/submitqueue/stovepipe/core/messagequeue/protopbb\x06proto3" var ( diff --git a/stovepipe/entity/issue_tracking_state.go b/stovepipe/entity/issue_tracking_state.go index 453e49e25..ded157e72 100644 --- a/stovepipe/entity/issue_tracking_state.go +++ b/stovepipe/entity/issue_tracking_state.go @@ -28,7 +28,7 @@ const ( IssueTrackingPhaseUnknown IssueTrackingPhase = "" // IssueTrackingPhaseRefreshing means reported build issues are being opened or kept open. IssueTrackingPhaseRefreshing IssueTrackingPhase = "refreshing" - // IssueTrackingPhaseRefreshed means a refresh pass completed; later refresh notifications remain eligible. + // IssueTrackingPhaseRefreshed means a refresh pass completed; later refresh passes remain eligible. IssueTrackingPhaseRefreshed IssueTrackingPhase = "refreshed" // IssueTrackingPhaseFinalizing means existing issues are being reconciled after all reported issues were applied. IssueTrackingPhaseFinalizing IssueTrackingPhase = "finalizing" @@ -42,18 +42,18 @@ const ( // IssueTrackingState is the queue-scoped source-order boundary and resumable reconciliation state. type IssueTrackingState struct { - // RequestID is the newest accepted source request, compared by ingest order. + // RequestID identifies the current or most recently completed validation. RequestID string `json:"request_id"` - // BuildID identifies the source build whose pass is underway or most recently completed. + // BuildID identifies the source build; it can be empty when abandoning before association. BuildID string `json:"build_id"` - // NotificationID identifies the message that owns this reconciliation pass. + // NotificationID identifies the tracking job owning this pass. NotificationID string `json:"notification_id"` // CursorTaskID identifies the current issue's task; active passes revisit it on continuation. // Empty with an empty CursorUnitID starts the phase's range. CursorTaskID string `json:"cursor_task_id"` // CursorUnitID identifies the unit at the cursor; empty is task-wide. CursorUnitID string `json:"cursor_unit_id"` - // Finalize identifies a terminal reconciliation pass rather than an ordinary refresh. + // Finalize identifies a terminal pass or a whole-validation abandonment. Finalize bool `json:"finalize"` // HasTaskWideIssues records whether this pass has encountered a task-wide issue. HasTaskWideIssues bool `json:"has_task_wide_issues"` @@ -69,7 +69,17 @@ func (s IssueTrackingState) Validate() error { return fmt.Errorf("invalid issue tracking request ID: %w", err) } if s.BuildID == "" { - return fmt.Errorf("issue tracking build ID must not be empty") + switch s.Phase { + case IssueTrackingPhaseAbandoning, IssueTrackingPhaseAbandoned: + if !s.Finalize { + return fmt.Errorf("abandoning a request without a build requires terminal intent") + } + if s.CursorTaskID != "" { + return fmt.Errorf("issue tracking without a build cannot have an issue cursor") + } + default: + return fmt.Errorf("issue tracking build ID must not be empty") + } } if s.NotificationID == "" { return fmt.Errorf("issue tracking notification ID must not be empty") diff --git a/stovepipe/entity/issue_tracking_state_test.go b/stovepipe/entity/issue_tracking_state_test.go index 245cc2512..c37f407be 100644 --- a/stovepipe/entity/issue_tracking_state_test.go +++ b/stovepipe/entity/issue_tracking_state_test.go @@ -56,3 +56,22 @@ func TestIssueTrackingState(t *testing.T) { }) } } + +func TestUnassociatedTrackingCanOnlyBeAbandonedWithoutAnIssueCursor(t *testing.T) { + for _, phase := range []IssueTrackingPhase{IssueTrackingPhaseRefreshed, IssueTrackingPhaseRefreshing, IssueTrackingPhaseFinalizing, IssueTrackingPhaseFinalized, IssueTrackingPhaseAbandoning, IssueTrackingPhaseAbandoned} { + t.Run(string(phase), func(t *testing.T) { + state := IssueTrackingState{RequestID: "10", NotificationID: "10/issues", Phase: phase, Finalize: true, Version: 1} + if phase == IssueTrackingPhaseAbandoning || phase == IssueTrackingPhaseAbandoned { + require.NoError(t, state.Validate()) + state.Finalize = false + require.Error(t, state.Validate()) + state.Finalize = true + state.CursorUnitID = "unit" + require.Error(t, state.Validate()) + state.CursorUnitID = "" + state.CursorTaskID = "task" + } + require.Error(t, state.Validate()) + }) + } +} diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index 4a335c196..c194e43bb 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -80,9 +80,9 @@ History persistence does not alter issue state or validate transitions against i ## Issue reconciliation progress -`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing, while `HasTaskWideIssues` records whether a task-wide issue has been encountered in that pass. Once all build issues have been applied, a terminal pass with task-wide failures finishes without closing; otherwise it resets the cursor and finalizes against issue states. These fields and the cursor are saved together under the same version guard. Notification identity lets the controller distinguish its own continuation from helping another pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. +`IssueTrackingStore` retains one queue-scoped operational `IssueTrackingState`, independently of open-issue membership and user-facing summaries. It records the source request and build, the refresh or finalization phase, the owning notification, one task/unit cursor, and its optimistic version. The zero cursor starts the current phase's range. Every active pass revisits the cursor issue before scanning later keys, including when the previous delivery already finished that issue. `Finalize` retains the terminal notification intent during refreshing, while `HasTaskWideIssues` records whether a task-wide issue has been encountered in that pass. Once all build issues have been applied, a terminal pass with task-wide failures finishes without closing; otherwise it resets the cursor and finalizes against issue states. These fields and the cursor are saved together under the same version guard. Notification identity retains the tracking job that owns the pass. The record remains after all issues close to preserve the source-order boundary. The MySQL table is keyed only by queue; creates, reads, and updates use complete primary keys with no secondary indexes or cross-table transaction. -The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. A later controller will own acceptance, resumption of cursor work, and completion; no runtime writer is enabled here. +The store does not decide source ordering, apply issue transitions, or advance versions. Its caller supplies the guard version and replacement version. The controller owns acceptance, resumption of cursor work, and completion. Recovery uses the same record's `abandoning` and `abandoned` phases; no additional schema fields or storage operations are required. The source, notification, cursor, and terminal intent remain present after abandonment. See [issue tracking recovery](../../controller/issuetrack/README.md#recovery-and-activation) for the conditional-write ordering that protects the current cursor before newer work is admitted. diff --git a/test/e2e/stovepipe/issuetrack_test.go b/test/e2e/stovepipe/issuetrack_test.go index 780e23b02..6d6b17bcf 100644 --- a/test/e2e/stovepipe/issuetrack_test.go +++ b/test/e2e/stovepipe/issuetrack_test.go @@ -68,16 +68,16 @@ func (s *StovepipeE2ESuite) TestIssueTracking_FinalizesAfterAnalysis() { require.Empty(t, issues) } -func (s *StovepipeE2ESuite) TestIssueTracking_DLQAbandonsPrematureFinalNotification() { +func (s *StovepipeE2ESuite) TestIssueTracking_DLQAbandonsInvalidBuildAssociation() { t := s.T() const queue = issueTrackingDLQQueue store, err := s.appStorage.For(queue) require.NoError(t, err) request := entity.Request{ ID: "1", Queue: queue, URI: "git://" + queue + "/HEAD", - State: entity.RequestStateProcessing, BuildStrategy: entity.BuildStrategyFull, Version: 1, + State: entity.RequestStateProcessing, BuildStrategy: entity.BuildStrategyFull, BuildID: "running-build", Version: 1, } - build := entity.Build{ID: "running-build", RequestID: request.ID, Status: entity.BuildStatusRunning, Version: 1} + build := entity.Build{ID: "running-build", RequestID: "another-request", Status: entity.BuildStatusRunning, Version: 1} require.NoError(t, store.GetRequestStore().Create(s.ctx, request)) require.NoError(t, store.GetBuildStore().Create(s.ctx, build)) require.NoError(t, store.GetQueueStore().Create(s.ctx, entity.Queue{Name: queue, InFlightCount: 1, Version: 1})) @@ -91,9 +91,9 @@ func (s *StovepipeE2ESuite) TestIssueTracking_DLQAbandonsPrematureFinalNotificat {Key: stovepipemq.TopicKeyIssueTrack, Name: "issuetrack", Queue: mq}, }) require.NoError(t, err) - payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{QueueName: queue, BuildId: build.ID, Finalize: true}) + payload, err := stovepipemq.Marshal(&stovepipemq.IssueTrack{QueueName: queue, RequestId: request.ID}) require.NoError(t, err) - notificationID := publish.IntentID(build.ID, "premature-final") + notificationID := publish.IntentID(request.ID, "issues") require.NoError(t, publish.Message(entityqueue.WithQueueName(s.ctx, queue), registry, stovepipemq.TopicKeyIssueTrack, publish.MessageParams{ Tenant: queue, ID: notificationID, Payload: payload, PartitionKey: queue, })) diff --git a/test/integration/extension/messagequeue/mysql/queue_test.go b/test/integration/extension/messagequeue/mysql/queue_test.go index af2e6e750..d8a07e07a 100644 --- a/test/integration/extension/messagequeue/mysql/queue_test.go +++ b/test/integration/extension/messagequeue/mysql/queue_test.go @@ -2899,3 +2899,31 @@ func (s *SQLQueueIntegrationSuite) TestWatermarkAdvancesContiguously() { t.Logf("Verified: watermark advances contiguously with out-of-order acks") } + +func (s *SQLQueueIntegrationSuite) TestSingleMessageBatchKeepsTheUnacknowledgedHead() { + t := s.T() + signalCh := make(chan queueMySQL.HookSignal, 100) + q, err := queueMySQL.NewQueue(s.testQueueParams(t, func(p *queueMySQL.Params) { p.OnSignal = signalCh })) + require.NoError(t, err) + defer q.Close() + const topic, partition = "single_head_topic", "single-head" + for _, id := range []string{"first", "second"} { + require.NoError(t, q.Publisher().Publish(s.ctx, topic, testMessage(id, []byte(id), partition, nil))) + } + config := testSubConfig("single-head-worker", "single-head-group") + config.BatchSize = 1 + deliveries, err := q.Subscriber().Subscribe(s.ctx, topic, config) + require.NoError(t, err) + first := receive(t, deliveries) + require.Equal(t, "first", first.Message().ID) + assertNoDelivery(t, deliveries, signalCh, queueMySQL.SignalDeliveryCheck, 3) + require.NoError(t, first.Postpone(s.ctx, 0)) + resumed := receive(t, deliveries) + require.Equal(t, "first", resumed.Message().ID) + require.Equal(t, 1, resumed.Attempt()) + assertNoDelivery(t, deliveries, signalCh, queueMySQL.SignalDeliveryCheck, 3) + require.NoError(t, resumed.Ack(s.ctx)) + second := receive(t, deliveries) + require.Equal(t, "second", second.Message().ID) + require.NoError(t, second.Ack(s.ctx)) +}