amadan

carlos / platform Public

e2e: TestEveryBoundaryRecovers pins each faulted write's announcement before the next write

notified_seq must reach the faulted write's landed record_seq
(expect_seq + 1) before the job's next record write is attempted, and by
the end; the request at a bump-point crash, already confirmed, is now
captured too. A mutation without the catch-up bump turns the queue,
restage and precheck boundaries red.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude abb33a4de7da0fa2d716d2ac3dd5e75e228c4f1f parent f2b4b38
2 files changed, +38 −7
  • internal/e2e/selfserve/job_boundary_test.go +29 −5
  • internal/e2e/selfserve/job_world_test.go +9 −2
diff --git a/internal/e2e/selfserve/job_boundary_test.go b/internal/e2e/selfserve/job_boundary_test.go
index 348624efd..117769a62 100644
--- a/internal/e2e/selfserve/job_boundary_test.go
+++ b/internal/e2e/selfserve/job_boundary_test.go
@@ -3,6 +3,7 @@
package selfserve
import (
+ "fmt"
"strings"
"testing"
"time"
@@ -151,6 +152,7 @@ func everyBoundary(t *testing.T, b jobBoundary, f jobFault, second bool) {
return true
}
var nonce string
+ var expect uint64
if f.put != putRun {
fired := false
w.fs.setPut(func(key string, body []byte) putFault {
@@ -158,7 +160,7 @@ func everyBoundary(t *testing.T, b jobBoundary, f jobFault, second bool) {
if fired || !inFlight(r) {
return putRun
}
- fired, nonce = true, r.Request.Nonce
+ fired, nonce, expect = true, r.Request.Nonce, r.Request.ExpectSeq
return f.put
})
}
@@ -172,8 +174,30 @@ func everyBoundary(t *testing.T, b jobBoundary, f jobFault, second bool) {
if f.point == "" {
w.fs.setPut(nil)
}
- if c := w.crashed; c != nil && inFlight(c) {
- nonce = c.Request.Nonce
+ if c := w.crashed; c != nil && c.Request.Kind == b.kind && c.Request.Phase == b.phase && !c.RequestIssued.IsZero() && !c.Request.Consumed {
+ // The request at the crash, confirmed already at the bump points.
+ nonce, expect = c.Request.Nonce, c.Request.ExpectSeq
+ }
+ // The faulted write lands at expect_seq + 1: notified_seq reaches it
+ // before the job's next record write is attempted.
+ var unannounced []string
+ if nonce != "" {
+ landed := expect + 1
+ checked := false
+ w.observe = func(point string, r *console.SelfServeMove) {
+ if checked || !strings.HasSuffix(point, ":write") || r.Request.Nonce == nonce {
+ return
+ }
+ checked = true
+ if r.NotifiedSeq < landed {
+ unannounced = append(unannounced, fmt.Sprintf("%s with notified_seq %d below the faulted write's %d", point, r.NotifiedSeq, landed))
+ }
+ }
+ defer func() {
+ if final := w.row(row.ID); final.NotifiedSeq < landed {
+ t.Errorf("notified_seq %d, below the faulted write's landed record_seq %d", final.NotifiedSeq, landed)
+ }
+ }()
}
if f.point != "" {
w.fs.setPut(nil)
@@ -246,8 +270,8 @@ func everyBoundary(t *testing.T, b jobBoundary, f jobFault, second bool) {
// finaliser can share one tick; every other must be seen.
t.Fatalf("request %s was never confirmed (final %s)", nonce, describe(final))
}
- if final.NotifiedSeq == 0 {
- t.Fatalf("notified_seq is 0 after the landed write")
+ if len(unannounced) > 0 {
+ t.Fatalf("the next write came before the faulted write was announced: %v", unannounced)
}
}
if w.fs.bumpCount() <= bumpsBefore {
diff --git a/internal/e2e/selfserve/job_world_test.go b/internal/e2e/selfserve/job_world_test.go
index 3c43920ba..c71bbe9ed 100644
--- a/internal/e2e/selfserve/job_world_test.go
+++ b/internal/e2e/selfserve/job_world_test.go
@@ -172,8 +172,10 @@ type jobWorld struct {
crashWhen func(point string, r *console.SelfServeMove) bool
crashed *console.SelfServeMove
snaps []console.SelfServeMove
- // on runs a witness's action at a crash point (once).
- on map[string]func()
+ // on runs a witness's action at a crash point (once); observe, when
+ // set, sees every crash point with the watched row as it is then.
+ on map[string]func()
+ observe func(point string, r *console.SelfServeMove)
// failKinds are the request kinds whose record writes fail (the
// write's own Put, between its write and written points; a fence or
// any other write still runs).
@@ -251,6 +253,11 @@ func (w *jobWorld) fault(point string) {
w.crash = ""
console.TestHooks.Crash(point)
}
+ if w.observe != nil && w.watch != "" {
+ if r, err := w.cst.SelfServeMove(w.watch); err == nil && r != nil {
+ w.observe(point, r)
+ }
+ }
if w.crashWhen != nil && w.watch != "" {
r, err := w.cst.SelfServeMove(w.watch)
if err == nil && r != nil && w.crashWhen(point, r) {