| index 8c46b0fd6..a3d6e1c38 100644 |
| --- a/internal/hostclaim/generation_test.go |
| +++ b/internal/hostclaim/generation_test.go |
| @@ -7,6 +7,7 @@ import ( |
| "fmt" |
| "reflect" |
| "testing" |
| + "time" |
| |
| "github.com/carlosframework/platform/internal/release" |
| ) |
| @@ -16,7 +17,7 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) { |
| s := release.NewDirStore(t.TempDir()) |
| o := servingClaim(t, ctx, s) |
| stamp := o.Claim.Holder.Stamp |
| - if o.Claim.Holder.Session != 0 { |
| + if o.Claim.Holder.Session != 0 || o.Claim.Holder.WriterBox != "" { |
| t.Fatalf("legacy session = %d, want zero", o.Claim.Holder.Session) |
| } |
| raw, err := json.Marshal(o.Claim.Holder) |
| @@ -27,15 +28,17 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) { |
| if err := json.Unmarshal(raw, &fields); err != nil { |
| t.Fatal(err) |
| } |
| - if _, ok := fields["session"]; ok { |
| - t.Fatal("zero session must be omitted") |
| + for _, name := range []string{"session", "writer_box"} { |
| + if _, ok := fields[name]; ok { |
| + t.Fatalf("zero %s must be omitted", name) |
| + } |
| } |
| - o, err = Update(ctx, s, o, AdmitWriter(0, "", "adi-knitting", "cold", "gen_a")) |
| + o, err = Update(ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a", PredecessorCertified: true})) |
| if err != nil { |
| t.Fatal(err) |
| } |
| cold := o.Claim.Gen |
| - if o.Claim.Holder.Generation != "gen_a" || o.Claim.Holder.Stamp != cold || o.Claim.Holder.Session != cold || cold <= stamp { |
| + if o.Claim.Holder.WriterBox != "adi-knitting" || o.Claim.Holder.Generation != "gen_a" || o.Claim.Holder.Stamp != cold || o.Claim.Holder.Session != cold || cold <= stamp { |
| t.Fatalf("cold holder = %+v, gen = %d", o.Claim.Holder, cold) |
| } |
| first := fmt.Sprintf(`{"session":%d,"box":"adi-knitting","generation":"gen_a","stamp":%d,"mode":"cold","predecessor":{"session":0,"box":"","generation":"","stamp":%d}}`, cold, cold, stamp) |
| @@ -47,17 +50,17 @@ func TestAdmitWriterWarmColdJournal(t *testing.T) { |
| if err != nil { |
| t.Fatal(err) |
| } |
| - o, err = Update(ctx, s, o, AdmitWriter(cold, "gen_a", "boxT", "warm", "")) |
| + o, err = Update(ctx, s, o, AdmitWriter(Admission{ObservedSession: cold, ObservedGeneration: "gen_a", Box: "boxT", Mode: "warm", PredecessorCertified: true})) |
| if err != nil { |
| t.Fatal(err) |
| } |
| warm := o.Claim.Gen |
| - if o.Claim.Holder.Session != warm || warm <= cold || o.Claim.Holder.Generation != "gen_a" || o.Claim.Holder.Stamp != cold { |
| + if o.Claim.Holder.WriterBox != "boxT" || o.Claim.Holder.Session != warm || warm <= cold || o.Claim.Holder.Generation != "gen_a" || o.Claim.Holder.Stamp != cold { |
| t.Fatalf("warm holder = %+v, gen = %d", o.Claim.Holder, warm) |
| } |
| second := fmt.Sprintf(`{"session":%d,"box":"boxT","generation":"gen_a","stamp":%d,"mode":"warm","predecessor":{"session":%d,"box":"adi-knitting","generation":"gen_a","stamp":%d}}`, warm, cold, cold, cold) |
| assertWriterJournal(t, o.Claim, "["+first+","+second+"]") |
| - o, err = Update(ctx, s, o, AdmitWriter(warm, "gen_a", "boxT", "cold", "gen_b")) |
| + o, err = Update(ctx, s, o, AdmitWriter(Admission{ObservedSession: warm, ObservedGeneration: "gen_a", Box: "boxT", Mode: "cold", NewGeneration: "gen_b", PredecessorCertified: true})) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -90,7 +93,7 @@ func assertWriterJournal(t *testing.T, c *Claim, want string) { |
| } |
| |
| func TestAdmitWriterPreservesOperation(t *testing.T) { |
| - for _, kind := range []string{KindServing, KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing, KindMoving, KindLanding, KindReverting, KindRestoring, KindPushing} { |
| + for _, kind := range []string{KindServing, KindCreating, KindMoving, KindLanding, KindReverting, KindRestoring, KindPushing} { |
| for _, mode := range []string{"warm", "cold"} { |
| t.Run(kind+"/"+mode, func(t *testing.T) { |
| ctx := context.Background() |
| @@ -110,7 +113,7 @@ func TestAdmitWriterPreservesOperation(t *testing.T) { |
| if mode == "cold" { |
| generation = "gen_a" |
| } |
| - next, err := Update(ctx, s, o, AdmitWriter(0, "", "boxT", mode, generation)) |
| + next, err := Update(ctx, s, o, AdmitWriter(Admission{ObservedOpID: "m1", ObservedPhase: "probe", Box: "boxT", Mode: mode, NewGeneration: generation, PredecessorCertified: true})) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -119,6 +122,7 @@ func TestAdmitWriterPreservesOperation(t *testing.T) { |
| want.Holder = &holder |
| want.Gen++ |
| want.Holder.Session = want.Gen |
| + want.Holder.WriterBox = "boxT" |
| if mode == "cold" { |
| want.Holder.Generation = generation |
| want.Holder.Stamp = want.Gen |
| @@ -133,7 +137,7 @@ func TestAdmitWriterPreservesOperation(t *testing.T) { |
| } |
| |
| func TestAdmitWriterRefusals(t *testing.T) { |
| - for _, name := range []string{"no holder", "no executor", "non-executor", "stale session", "stale generation", "warm with generation", "cold without generation", "cold same generation", "unknown mode", "empty mode", "malformed journal"} { |
| + for _, name := range []string{"no holder", "no executor", "non-executor", "stale session", "stale generation", "warm with generation", "cold without generation", "cold same generation", "unknown mode", "empty mode", "malformed journal", "warm without certificate", "stale op", "stale phase", "deleting", "parking", "purging", "placing"} { |
| t.Run(name, func(t *testing.T) { |
| ctx := context.Background() |
| s := release.NewDirStore(t.TempDir()) |
| @@ -147,6 +151,8 @@ func TestAdmitWriterRefusals(t *testing.T) { |
| c.Op = Op{} |
| case "no executor": |
| c.Op.Exec = "" |
| + case "deleting", "parking", "purging", "placing": |
| + c.Op.Kind, c.Op.ID = name, "op1" |
| case "malformed journal": |
| c.Op.Journal = map[string]string{"writers": "not-json"} |
| } |
| @@ -176,7 +182,16 @@ func TestAdmitWriterRefusals(t *testing.T) { |
| case "empty mode": |
| mode = "" |
| } |
| - if _, err := Update(ctx, s, o, AdmitWriter(session, generation, box, mode, newGeneration)); !errors.Is(err, ErrWrongState) { |
| + in := Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase, ObservedSession: session, ObservedGeneration: generation, Box: box, Mode: mode, NewGeneration: newGeneration, PredecessorCertified: true} |
| + switch name { |
| + case "warm without certificate": |
| + in.Mode, in.NewGeneration, in.PredecessorCertified = "warm", "", false |
| + case "stale op": |
| + in.ObservedOpID = "old" |
| + case "stale phase": |
| + in.ObservedPhase = "old" |
| + } |
| + if _, err := Update(ctx, s, o, AdmitWriter(in)); !errors.Is(err, ErrWrongState) { |
| t.Fatalf("admit writer = %v, want ErrWrongState", err) |
| } |
| stored, err := Read(ctx, s, testHost) |
| @@ -189,3 +204,145 @@ func TestAdmitWriterRefusals(t *testing.T) { |
| }) |
| } |
| } |
| + |
| +func TestAdmitWriterCompetingAdmissions(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + before := servingClaim(t, ctx, s) |
| + in := Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a"} |
| + first := AdmitWriter(in) |
| + in.NewGeneration = "gen_b" |
| + second := AdmitWriter(in) |
| + admitted, err := Update(ctx, s, before, first) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := Update(ctx, s, before, second); !errors.Is(err, ErrConflict) { |
| + t.Fatalf("competing admission = %v, want ErrConflict", err) |
| + } |
| + current := assertAdmissionStored(t, ctx, s, admitted) |
| + if _, err := Update(ctx, s, current, second); !errors.Is(err, ErrWrongState) { |
| + t.Fatalf("stale evidence after reread = %v, want ErrWrongState", err) |
| + } |
| + assertAdmissionStored(t, ctx, s, admitted) |
| +} |
| + |
| +func TestAdmitWriterLostResponse(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + before := servingClaim(t, ctx, s) |
| + admit := AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a"}) |
| + admitted, err := Update(ctx, s, before, admit) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := Update(ctx, s, before, admit); !errors.Is(err, ErrConflict) { |
| + t.Fatalf("retry after lost response = %v, want ErrConflict", err) |
| + } |
| + assertAdmissionStored(t, ctx, s, admitted) |
| +} |
| + |
| +func TestAdmitWriterBoundToOperationAndPhase(t *testing.T) { |
| + for _, change := range []string{"operation", "phase", "delete"} { |
| + t.Run(change, func(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + o := servingClaim(t, ctx, s) |
| + stale := AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a"}) |
| + var err error |
| + if change == "delete" { |
| + o, err = Update(ctx, s, o, BeginDelete("d1", "console", "etag", time.Now(), nil, nil)) |
| + } else { |
| + o, err = Update(ctx, s, o, func(c *Claim) error { |
| + if change == "operation" { |
| + c.Op.Kind, c.Op.ID = KindMoving, "m1" |
| + } else { |
| + c.Op.Phase = "probe" |
| + } |
| + return nil |
| + }) |
| + } |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := Update(ctx, s, o, stale); !errors.Is(err, ErrWrongState) { |
| + t.Fatalf("admission after %s changed = %v, want ErrWrongState", change, err) |
| + } |
| + assertAdmissionStored(t, ctx, s, o) |
| + if change == "delete" { |
| + fresh := AdmitWriter(Admission{ObservedOpID: "d1", Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a"}) |
| + if _, err := Update(ctx, s, o, fresh); !errors.Is(err, ErrWrongState) { |
| + t.Fatalf("deleting admission = %v", err) |
| + } |
| + assertAdmissionStored(t, ctx, s, o) |
| + id := 0 |
| + o, err = Update(ctx, s, o, ToParking("d1", nil, nil, func() string { id++; return fmt.Sprintf("o%d", id) })) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := Update(ctx, s, o, fresh); !errors.Is(err, ErrWrongState) { |
| + t.Fatalf("parking admission = %v", err) |
| + } |
| + assertAdmissionStored(t, ctx, s, o) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestAdmitWriterSurvivesOperationReplacement(t *testing.T) { |
| + for _, transition := range []string{"delete", "move"} { |
| + t.Run(transition, func(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + o := servingClaim(t, ctx, s) |
| + o, err := Update(ctx, s, o, AdmitWriter(Admission{Box: "adi-knitting", Mode: "cold", NewGeneration: "gen_a"})) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + writer := *o.Claim.Holder |
| + if writer.WriterBox != "adi-knitting" || writer.Session != o.Claim.Gen { |
| + t.Fatalf("admitted holder = %+v", writer) |
| + } |
| + if transition == "delete" { |
| + o, err = Update(ctx, s, o, BeginDelete("d1", "console", "etag", time.Now(), nil, nil)) |
| + } else { |
| + o, err = Update(ctx, s, o, func(c *Claim) error { |
| + c.Op = Op{Kind: KindMoving, ID: "m1", Phase: "probe", Exec: "boxT"} |
| + return nil |
| + }) |
| + } |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + o = assertAdmissionStored(t, ctx, s, o) |
| + if !reflect.DeepEqual(*o.Claim.Holder, writer) { |
| + t.Fatalf("%s changed writer: %+v", transition, o.Claim.Holder) |
| + } |
| + if transition == "delete" { |
| + return |
| + } |
| + o, err = Update(ctx, s, o, AdmitWriter(Admission{ObservedOpID: "m1", ObservedPhase: "probe", ObservedSession: writer.Session, ObservedGeneration: writer.Generation, Box: "boxT", Mode: "cold", NewGeneration: "gen_b", RemoteMaxTXID: 42})) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := fmt.Sprintf(`[{"session":%d,"box":"boxT","generation":"gen_b","stamp":%d,"mode":"cold","predecessor":{"session":%d,"box":"adi-knitting","generation":"gen_a","stamp":%d},"writes_lost":{"box":"adi-knitting","generation":"gen_a","session":%d,"remote_max_txid":42}}]`, o.Claim.Gen, o.Claim.Gen, writer.Session, writer.Stamp, writer.Session) |
| + assertWriterJournal(t, o.Claim, want) |
| + if o.Claim.Holder.WriterBox != "boxT" { |
| + t.Fatalf("new writer = %q", o.Claim.Holder.WriterBox) |
| + } |
| + assertAdmissionStored(t, ctx, s, o) |
| + }) |
| + } |
| +} |
| + |
| +func assertAdmissionStored(t *testing.T, ctx context.Context, s release.Store, want Observed) Observed { |
| + t.Helper() |
| + got, err := Read(ctx, s, testHost) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !reflect.DeepEqual(got, want) { |
| + t.Fatalf("stored claim or ETag changed: got %+v, want %+v", got, want) |
| + } |
| + return got |
| +} |