fix(storage): classify version mismatch as retryable

doc/rfc/stovepipe/steps/buildsignal.md

@@ -47,7 +47,7 @@ For a delivery carrying build id `B`:
      continue with the STORED status as authoritative (see Edge cases).
    - otherwise persist via BuildStore.Update(ctx, Build{...Status: status}, oldVersion, newVersion):
      - newVersion = Build.Version + 1; assign Build.Version = newVersion only on success.
-     - ErrVersionMismatch -> retryable (a concurrent writer moved the row; reload and re-check).
+     - ErrVersionMismatch -> return with its declaration-level retryable classification (a concurrent writer moved the row; reload and re-check).
      - with the write-once rule, accepted -> running -> {succeeded|failed|cancelled} is monotonic
        by mechanism, not by assumption about the backend.
 
@@ -66,7 +66,7 @@ For a delivery carrying build id `B`:
 
 **Why `record` hears only terminal signals**: `record` has no non-terminal work — by its own contract a non-terminal signal would be a pure no-op — and step 7 already branches on terminality to decide whether to keep polling, so gating the publish costs nothing and spares `record` a no-op delivery on every poll tick of every running build. Crash-safety is unaffected: a crash between the terminal `Update` and the publish redelivers the message; step 5 re-polls (the runner reports the same terminal status), step 6 no-ops, step 7 publishes. This is a deliberate divergence from SubmitQueue, whose buildsignal republishes to `speculate` on every tick — sound there because speculate is a state machine that may act on any signal; stovepipe has no such consumer.
 
-**Why step 6 guards on status and makes terminal write-once**: an unchanged status skips the CAS write entirely, so a long build being polled every couple of seconds doesn't churn `Build.Version` on every tick — the version only advances on a real state transition. The write-once rule exists because CAS alone cannot provide it: optimistic locking defends against *concurrent* writers, but a later delivery that polls a flaky backend and sees a different terminal status would CAS cleanly against the current version and overwrite (see Edge cases). A given `Build` has a single poll partition (see [Partitioning](doc/rfc/stovepipe/steps/build.md#partitioning)), so the only writer racing the CAS is a redelivery of the same message (e.g. after a lapsed visibility lease); `ErrVersionMismatch` there is handled as retryable and converges.
+**Why step 6 guards on status and makes terminal write-once**: an unchanged status skips the CAS write entirely, so a long build being polled every couple of seconds doesn't churn `Build.Version` on every tick — the version only advances on a real state transition. The write-once rule exists because CAS alone cannot provide it: optimistic locking defends against *concurrent* writers, but a later delivery that polls a flaky backend and sees a different terminal status would CAS cleanly against the current version and overwrite (see Edge cases). A given `Build` has a single poll partition (see [Partitioning](doc/rfc/stovepipe/steps/build.md#partitioning)), so the only writer racing the CAS is a redelivery of the same message (e.g. after a lapsed visibility lease); `ErrVersionMismatch` carries a retryable classification and converges on redelivery.
 
 ## Status: shaped like SubmitQueue's, not shared code
 
@@ -101,7 +101,7 @@ Per `platform/errs`'s non-retryable-by-default rule (see [platform/errs/README.m
 | Failure | Disposition | Why |
 |---|---|---|
 | `Status` call | raw error; classifier decides | Deliberately left open rather than fixed either way — runner timeout/connection is transient, "runner not deployed for this queue" is not, and only a backend classifier can tell them apart. |
-| `Update` CAS conflict (`ErrVersionMismatch`) | retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. |
+| `Update` CAS conflict (`ErrVersionMismatch`) | declaration-level retryable | A concurrent (redelivered) writer moved the row; reload and re-check converges. |
 | `PublishAfter` re-poll | retryable | The poll heartbeat; it runs only after status/persist/record all succeeded, so a transient enqueue blip is worth replaying to `MaxAttempts` before dead-lettering. |
 
 `Build`/`Request` not found (`storage.ErrNotFound`) are **not** in this table: storage is required to be read-after-write consistent (see [storage README](stovepipe/extension/storage/README.md)), so a miss here is already the correct default (non-retryable, straight to DLQ) rather than a departure worth overriding.

platform/errs/README.md

@@ -135,11 +135,11 @@ The controller-override path is for the rare case where the controller has certa
 
 In particular, **do not reach for `NewRetryableError` just because replaying the message would be convenient.** A failed queue publish, a failed enqueue, a "the hand-off that keeps this alive" step — these are *not* a license to mark the error retryable. Whether such a failure is transient is exactly what a classifier exists to decide: a transport-level classifier wraps genuine connection/timeout blips as retryable, while a malformed-request or permission failure stays non-retryable and dead-letters instead of replaying forever. Blanket `NewRetryableError` on a publish path defeats that and turns every permanent failure into an infinite retry loop.
 
-## Extensions Return Plain Go Errors
+## Extensions Return Go Errors
 
-Extension interfaces (`MergeChecker`, `Storage`, `Publisher`) return standard `error` values. They may define their own domain-specific sentinel errors (e.g. `storage.ErrNotFound`, `storage.ErrVersionMismatch`) but they do **not** classify errors as user or infra — that is the controller's (and the consumer's `ErrorProcessor`'s) job.
+Extension interfaces (`MergeChecker`, `Storage`, `Publisher`) return `error` values and may define domain-specific sentinels. Most sentinels remain unclassified because their meaning depends on the call site; for example, `storage.ErrNotFound` might be a user error in one controller and an infrastructure error in another. A sentinel whose classification is intrinsic in every context may carry that classification at its declaration. `storage.ErrVersionMismatch`, for example, is always a retryable infrastructure error because it reports a lost optimistic-concurrency race.
 
-This separation keeps extensions reusable across contexts. The same `storage.ErrNotFound` might be a user error in one controller (user requested a non-existent resource) and an infra error in another (expected record is missing).
+Controllers should return intrinsically classified sentinels without adding another framework wrapper. The declaration remains reusable across implementations while every caller observes the same classification.
 
 ## Error Chain Compatibility
 

stovepipe/controller/buildsignal/buildsignal.go

@@ -21,7 +21,6 @@ package buildsignal
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	"github.com/uber-go/tally"
@@ -185,9 +184,6 @@ func (c *Controller) reconcile(ctx context.Context, build entity.Build, status e
 	updated := build
 	updated.Status = status
 	if err := c.store.GetBuildStore().Update(ctx, updated, build.Version, newVersion); err != nil {
-		if errors.Is(err, storage.ErrVersionMismatch) {
-			return "", errs.NewRetryableError(fmt.Errorf("build %s version conflict: %w", build.ID, err))
-		}
 		return "", fmt.Errorf("failed to persist status for build %s: %w", build.ID, err)
 	}
 	return status, nil

stovepipe/extension/storage/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
@@ -11,5 +11,18 @@ go_library(
     ],
     importpath = "github.com/uber/submitqueue/stovepipe/extension/storage",
     visibility = ["//visibility:public"],
-    deps = ["//stovepipe/entity:go_default_library"],
+    deps = [
+        "//platform/errs:go_default_library",
+        "//stovepipe/entity:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["storage_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/errs:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+    ],
 )

stovepipe/extension/storage/README.md

@@ -1,6 +1,6 @@
 # Storage
 
-Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`). Implementations live under `extension/storage/<impl>/`. This is a separate contract from `submitqueue/extension/storage` — same shape and conventions by design, but its own interfaces and its own `ErrNotFound`/`ErrAlreadyExists`/`ErrVersionMismatch` sentinels, since Stovepipe and SubmitQueue are independent domains.
+Pluggable persistence interfaces for Stovepipe entities (`RequestStore`, `RequestURIStore`, `QueueStore`, `BuildStore`). Implementations live under `extension/storage/<impl>/`. This is a separate contract from `submitqueue/extension/storage` — same shape and conventions by design, but its own interfaces and its own `ErrNotFound`/`ErrAlreadyExists`/`ErrVersionMismatch` sentinels, since Stovepipe and SubmitQueue are independent domains. `ErrVersionMismatch` is declared as a retryable infrastructure error so callers can return it without reclassifying it.
 
 ## Optimistic locking contract
 

stovepipe/extension/storage/storage.go

@@ -19,6 +19,8 @@ package storage
 import (
 	"errors"
 	"fmt"
+
+	"github.com/uber/submitqueue/platform/errs"
 )
 
 // ErrNotFound is returned by storage implementations when the requested record is not found in the database.
@@ -39,8 +41,8 @@ var ErrAlreadyExists = errors.New("record already exists")
 
 // ErrVersionMismatch is returned by storage implementations when a conditional (CAS) update finds that
 // the stored version does not match the expected version. It backs optimistic locking, letting callers
-// retry or converge instead of overwriting a concurrent change.
-var ErrVersionMismatch = errors.New("version mismatch")
+// retry or converge instead of overwriting a concurrent change. It is intrinsically a retryable infrastructure error.
+var ErrVersionMismatch = errs.NewRetryableError(errors.New("version mismatch"))
 
 // Storage is a factory interface that aggregates all entity stores into a single injectable dependency.
 type Storage interface {

stovepipe/extension/storage/storage_test.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package storage
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/uber/submitqueue/platform/errs"
+)
+
+func TestErrVersionMismatchClassification(t *testing.T) {
+	err := fmt.Errorf("update request: %w", ErrVersionMismatch)
+
+	assert.ErrorIs(t, err, ErrVersionMismatch)
+	assert.True(t, errs.IsRetryable(err))
+	assert.False(t, errs.IsUserError(err))
+	assert.False(t, errs.IsDependencyError(err))
+}

submitqueue/extension/storage/BUILD.bazel

@@ -1,4 +1,4 @@
-load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
@@ -16,5 +16,18 @@ go_library(
     ],
     importpath = "github.com/uber/submitqueue/submitqueue/extension/storage",
     visibility = ["//visibility:public"],
-    deps = ["//submitqueue/entity:go_default_library"],
+    deps = [
+        "//platform/errs:go_default_library",
+        "//submitqueue/entity:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["storage_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//platform/errs:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+    ],
 )

submitqueue/extension/storage/README.md

@@ -4,7 +4,7 @@ Pluggable persistence interfaces for SubmitQueue entities (requests, batches, de
 
 ## Optimistic locking contract
 
-Entities that support concurrent mutation carry an `int32 Version` field. Updates are conditional on the version: the write only succeeds if the persisted version matches the caller's expected version. On mismatch, the implementation returns `storage.ErrVersionMismatch`.
+Entities that support concurrent mutation carry an `int32 Version` field. Updates are conditional on the version: the write only succeeds if the persisted version matches the caller's expected version. On mismatch, the implementation returns `storage.ErrVersionMismatch`, which is declared as a retryable infrastructure error so callers can return it without reclassifying it.
 
 **Version arithmetic is owned by the controller, not the store.** Update methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write):
 

submitqueue/extension/storage/storage.go

@@ -19,6 +19,8 @@ package storage
 import (
 	"errors"
 	"fmt"
+
+	"github.com/uber/submitqueue/platform/errs"
 )
 
 // ErrNotFound is returned by storage implementations when the requested record is not found in the database.
@@ -39,8 +41,8 @@ var ErrAlreadyExists = errors.New("record already exists")
 
 // ErrVersionMismatch is returned by storage implementations when the expected entity version does not match the current version of the object.
 // This is used to implement an optimistic locking mechanism, allowing multiple clients to update the same entity concurrently
-// and either retry or implement idempotent operations.
-var ErrVersionMismatch = errors.New("version mismatch")
+// and either retry or implement idempotent operations. It is intrinsically a retryable infrastructure error.
+var ErrVersionMismatch = errs.NewRetryableError(errors.New("version mismatch"))
 
 // Storage is a factory interface that aggregates all entity stores into a single injectable dependency.
 type Storage interface {

submitqueue/extension/storage/storage_test.go

@@ -0,0 +1,32 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package storage
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/uber/submitqueue/platform/errs"
+)
+
+func TestErrVersionMismatchClassification(t *testing.T) {
+	err := fmt.Errorf("update request: %w", ErrVersionMismatch)
+
+	assert.ErrorIs(t, err, ErrVersionMismatch)
+	assert.True(t, errs.IsRetryable(err))
+	assert.False(t, errs.IsUserError(err))
+	assert.False(t, errs.IsDependencyError(err))
+}

submitqueue/orchestrator/controller/cancel/cancel.go

@@ -46,10 +46,10 @@
 // Cancelling batch re-publishes to TopicKeySpeculate (a cheap no-op nudge
 // the speculate controller absorbs).
 //
-// Concurrent producers surface as storage.ErrVersionMismatch; the controller
-// returns the wrapped error as-is and relies on the base controller layer to
-// classify it as retryable so the next attempt sees the new state and takes
-// the other branch. storage.ErrNotFound on the initial Get (the start
+// Concurrent producers surface as the intrinsically retryable
+// storage.ErrVersionMismatch; the controller returns the wrapped error as-is
+// so the next attempt sees the new state and takes the other branch.
+// storage.ErrNotFound on the initial Get (the start
 // controller has not yet persisted the request) is returned as-is for the
 // same reason.
 package cancel
@@ -161,8 +161,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er
 // where the prior delivery already wrote Cancelling.
 //
 // storage.ErrVersionMismatch (a concurrent writer — most likely conclude
-// observing a batch transition) is returned as-is for the base controller to
-// classify and retry; the next attempt re-fetches and re-evaluates (it may now
+// observing a batch transition) is returned as-is; its declaration makes it
+// retryable, and the next attempt re-fetches and re-evaluates (it may now
 // be terminal, in which case the top-level terminal-check acks).
 func (c *Controller) markCancelling(ctx context.Context, request entity.Request) (entity.Request, error) {
 	if request.State == entity.RequestStateCancelling {
@@ -212,8 +212,8 @@ func (c *Controller) findActiveBatch(ctx context.Context, request entity.Request
 // that is not part of any active batch, and emits the RequestStatusCancelled log
 // entry. storage.ErrVersionMismatch here means a concurrent writer (typically
 // conclude after a racing batch terminal transition) advanced the request between
-// our mark-cancelling CAS and this terminal CAS — returned as-is for the base
-// controller to classify and retry; the next pass will observe the new state
+// our mark-cancelling CAS and this terminal CAS — returned as-is because the
+// sentinel is intrinsically retryable; the next pass will observe the new state
 // (likely terminal) and ack via the top-level terminal-check.
 func (c *Controller) cancelRequest(ctx context.Context, request entity.Request, reason string) error {
 	newVersion := request.Version + 1
@@ -268,8 +268,8 @@ func (c *Controller) cancelBatch(ctx context.Context, batch entity.Batch) error
 		if err := c.store.GetBatchStore().UpdateState(ctx, batch.ID, batch.Version, newVersion, entity.BatchStateCancelling); err != nil {
 			c.metricsScope.Counter("batch_update_errors").Inc(1)
 			// storage.ErrVersionMismatch here means the batch advanced concurrently
-			// (e.g. speculate / merge progressed). Returned as-is for the base
-			// controller to classify and retry; the re-fetch will see the new state
+			// (e.g. speculate / merge progressed). Returned as-is because the
+			// sentinel is intrinsically retryable; the re-fetch will see the new state
 			// and either short-circuit (already terminal) or attempt the transition
 			// again.
 			return fmt.Errorf("failed to mark batch %s as cancelling: %w", batch.ID, err)

submitqueue/orchestrator/controller/cancel/cancel_test.go

@@ -188,8 +188,8 @@ func TestProcess_AlreadyCancelling_SkipsMarkCancelling(t *testing.T) {
 
 // TestProcess_MarkCancellingVersionMismatch_Retryable covers the case where the
 // first CAS (mark-cancelling) loses to a concurrent writer. The underlying
-// storage.ErrVersionMismatch must be preserved in the error chain so the base
-// controller can classify it as retryable; the next pass re-fetches and
+// storage.ErrVersionMismatch must be preserved in the error chain so its
+// intrinsic retryable classification survives; the next pass re-fetches and
 // re-evaluates (possibly observing a terminal state and acking).
 func TestProcess_MarkCancellingVersionMismatch_Retryable(t *testing.T) {
 	ctrl := gomock.NewController(t)
@@ -342,8 +342,7 @@ func TestProcess_BatchAlreadyCancelling_RepublishesToSpeculate(t *testing.T) {
 // TestProcess_BatchIntentVersionMismatch_Retryable covers the case where the
 // intent CAS (mark batch Cancelling) loses to a concurrent batch state
 // transition (e.g. speculate just advanced it). storage.ErrVersionMismatch
-// must be preserved so the base controller can classify the failure as
-// retryable.
+// must be preserved so its intrinsic retryable classification survives.
 func TestProcess_BatchIntentVersionMismatch_Retryable(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	registry, _ := newRegistry(t, ctrl)

submitqueue/orchestrator/controller/conclude/conclude_test.go

@@ -317,7 +317,7 @@ func TestController_Process(t *testing.T) {
 			retryable: false,
 		},
 		{
-			name: "request store update failure returns error",
+			name: "request store version mismatch is retryable",
 			batch: entity.Batch{
 				ID:       "test-queue/batch/6",
 				Queue:    "test-queue",
@@ -347,7 +347,7 @@ func TestController_Process(t *testing.T) {
 				return mockStorage
 			},
 			wantErr:   true,
-			retryable: false,
+			retryable: true,
 		},
 		{
 			name: "empty contains list succeeds",