amadan
self-serve job: every attempt of one request writes the same bytes (fix round 2, item 4)
provision.Expect gains Stamp, the request's proposed stamp: when set, SetRecordMoveAt writes it as the move's updated_at instead of the attempt's clock. The job's bound write passes the persisted request's proposed stamp, so a replay after a failure or a restart sends the first attempt's body byte for byte. The operator doors' bound write (offlineApplyBound) passes no Stamp: it makes one attempt and never replays, and its updated_at stays the clock, so its answers are unchanged. Witness: TestQueueReplayIsByteIdentical (e2e; failures, a restart, the landing). RED: "attempt 1 of request rq_… differs from the first".
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+63
−3
internal/console/selfserve_apply.go+1 −1internal/e2e/selfserve/queue_replay_test.go+50 −0internal/provision/provision.go+12 −2
diff --git a/internal/console/selfserve_apply.go b/internal/console/selfserve_apply.go| index 87efbf6bf..8a67743ba 100644 |
| --- a/internal/console/selfserve_apply.go |
| +++ b/internal/console/selfserve_apply.go |
| @@ -319,7 +319,7 @@ func (s *Server) selfServeRun(ctx context.Context, t *leaseToken, row *SelfServe |
| TestHooks.point(pr.Kind + ":write") |
| wctx, cancel := s.selfServeWriteCtx(ctx, t) |
| _, werr := provision.SetRecordMoveAt(wctx, s.cfg.Releases, row.Account, row.App, row.Host, |
| - provision.Expect{Incarnation: pr.ExpectInc, Seq: pr.ExpectSeq}, mut(pr.Proposed, provision.Receipt{Nonce: pr.Nonce})) |
| + provision.Expect{Incarnation: pr.ExpectInc, Seq: pr.ExpectSeq, Stamp: pr.Proposed}, mut(pr.Proposed, provision.Receipt{Nonce: pr.Nonce})) |
| cancel() |
| lastErr, refused = werr, selfServeMutationRefused(werr) |
| if werr != nil && !refused { |
diff --git a/internal/e2e/selfserve/queue_replay_test.go b/internal/e2e/selfserve/queue_replay_test.go| index a36d1dce0..d9104f306 100644 |
| --- a/internal/e2e/selfserve/queue_replay_test.go |
| +++ b/internal/e2e/selfserve/queue_replay_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "regexp" |
| "strings" |
| "testing" |
| + "time" |
| "github.com/carlosframework/platform/internal/host" |
| ) |
| @@ -97,3 +98,52 @@ func TestQueueReplayKeepsItsStagingCoordinates(t *testing.T) { |
| }) |
| } |
| } |
| + |
| +// TestQueueReplayIsByteIdentical (fix round 2, item 4): every attempt of |
| +// one request writes the same bytes. The queue's writes fail, the console |
| +// restarts and replays, failing again, and then lands: every body sent |
| +// under the request's nonce, the landed one included, is byte for byte the |
| +// first (the move's updated_at is the request's proposed stamp, not the |
| +// attempt's clock). |
| +func TestQueueReplayIsByteIdentical(t *testing.T) { |
| + w := newJobWorld(t) |
| + nonceOf := regexp.MustCompile(`"last_request":"(rq_[0-9a-f]+)"`) |
| + bodies := map[string][]string{} |
| + failing := true |
| + w.fs.setPut(func(key string, body []byte) putFault { |
| + b := string(body) |
| + if !strings.Contains(b, `"offline-queued"`) { |
| + return putRun |
| + } |
| + if n := nonceOf.FindStringSubmatch(b); n != nil { |
| + bodies[n[1]] = append(bodies[n[1]], b) |
| + } |
| + if failing { |
| + return putFail |
| + } |
| + return putRun |
| + }) |
| + row := w.admit() |
| + w.watch = row.ID |
| + w.until(20, "a queue PUT sent", func() bool { return len(bodies) > 0 }) |
| + nonce := w.row(row.ID).Request.Nonce |
| + w.advance(2 * time.Second) // the replay's clock is later than the issue's |
| + w.tick() |
| + w.restart() |
| + w.advance(2 * time.Second) |
| + w.tick() |
| + failing = false |
| + w.consoleUntil(10, "queued", func() bool { return w.record().Move != nil }) |
| + got := bodies[nonce] |
| + if len(got) < 3 { |
| + t.Fatalf("%d attempts of the request captured, want the failures, the restart's replay and the landing", len(got)) |
| + } |
| + for i, b := range got { |
| + if b != got[0] { |
| + t.Fatalf("attempt %d of request %s differs from the first:\n first %s\n this %s", i, nonce, got[0], b) |
| + } |
| + } |
| + if len(bodies) != 1 { |
| + t.Fatalf("%d request identities sent, want one", len(bodies)) |
| + } |
| +} |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index 17a6db058..b8b42a317 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -1433,8 +1433,15 @@ func SetRecordMove(ctx context.Context, s release.Store, acct, app, host string, |
| } |
| // Expect is the record a bound write was decided on: its incarnation and |
| -// its record_seq as the caller observed them. |
| -type Expect struct{ Incarnation, Seq uint64 } |
| +// its record_seq as the caller observed them. Stamp, when set, is the |
| +// request's proposed stamp (the self-serve job's persisted request): the |
| +// write's move updated_at is it rather than the attempt's clock, so every |
| +// attempt of one request writes the same bytes (self-serve moves plan 2, |
| +// final review F10). Zero keeps the clock, as every other writer does. |
| +type Expect struct { |
| + Incarnation, Seq uint64 |
| + Stamp time.Time |
| +} |
| // SetRecordMoveAt is SetRecordMove bound to the record its caller observed: |
| // it reads the record (judged through JudgeFrom(ctx) when the context |
| @@ -1458,6 +1465,9 @@ func SetRecordMoveAt(ctx context.Context, s release.Store, acct, app, host strin |
| } |
| if rec.Move != nil { |
| rec.Move.UpdatedAt = time.Now().UTC() |
| + if !expect.Stamp.IsZero() { |
| + rec.Move.UpdatedAt = expect.Stamp.UTC() |
| + } |
| } |
| return nil |
| }, false, &expect) |