| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "strings" |
| 8 | "testing" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/hostclaim" |
| 11 | "github.com/carlosframework/platform/internal/moves" |
| 12 | "github.com/carlosframework/platform/internal/provision" |
| 13 | ) |
| 14 | |
| 15 | // Plan 6 Task 3 fix rounds 3 and 4: the recovery pass's move log writer |
| 16 | // (Recovery.MoveLogs) delivers the console's queued entries whatever their |
| 17 | // moves now are. The entry names a host, a move, the instance incarnation the |
| 18 | // move's evidence carried, and an entry name; the account and app come from |
| 19 | // the host's claim, whose holder must be that incarnation, and the move must |
| 20 | // be one its record names (live, or closed as its last). The name is one |
| 21 | // segment and the entry's identity; the entry is at most MoveLogEntryMax |
| 22 | // bytes and this move's and host's; the write is create-only, a replay of the |
| 23 | // same bytes a no-op and other bytes ErrMoveLogTaken. It answers only during |
| 24 | // its hook, inside a recovery pass. |
| 25 | func TestMoveLogWriterWritesOnlyVerifiedEntries(t *testing.T) { |
| 26 | w := newC7World(t, func(c *hostclaim.Claim, r *provision.Record, _ *AppAdmission) {}) |
| 27 | ctx := context.Background() |
| 28 | entry := func(op, host, id, actor string) []byte { |
| 29 | b, err := json.Marshal(moves.LogEvent{V: 1, Op: op, Host: host, Event: "x", Identity: id, Actor: actor, At: w.now}) |
| 30 | if err != nil { |
| 31 | t.Fatal(err) |
| 32 | } |
| 33 | return b |
| 34 | } |
| 35 | good := MoveLogEntry{Host: w.host, Op: "move", Incarnation: w.inc, Name: "done", Body: entry("move", w.host, "done", "console")} |
| 36 | with := func(f func(*MoveLogEntry)) MoveLogEntry { e := good; f(&e); return e } |
| 37 | big := moves.LogEvent{V: 1, Op: "move", Host: w.host, Event: "x", Identity: "done", Fields: map[string]string{"pad": strings.Repeat("a", MoveLogEntryMax)}} |
| 38 | bigBody, _ := json.Marshal(big) |
| 39 | var saved *MoveLogWriter |
| 40 | var created, again bool |
| 41 | got := map[string]error{} |
| 42 | err := w.control.RunRecovery(ctx, func(ctx context.Context) error { |
| 43 | return runMoveLogDrain(ctx, func(ctx context.Context, lw *MoveLogWriter) error { |
| 44 | saved = lw |
| 45 | var err error |
| 46 | if created, err = lw.WriteMoveLog(ctx, good); err != nil { |
| 47 | return err |
| 48 | } |
| 49 | if again, err = lw.WriteMoveLog(ctx, good); err != nil { |
| 50 | return err |
| 51 | } |
| 52 | for what, e := range map[string]MoveLogEntry{ |
| 53 | "other bytes": with(func(e *MoveLogEntry) { e.Body = entry("move", w.host, "done", "target") }), |
| 54 | "a stale incarnation (recreated under the same names)": with(func(e *MoveLogEntry) { e.Incarnation = strings.Repeat("ef", 32) }), |
| 55 | "a move its record does not name": with(func(e *MoveLogEntry) { e.Op = "forged"; e.Body = entry("forged", w.host, "done", "console") }), |
| 56 | "another move's entry": with(func(e *MoveLogEntry) { e.Body = entry("other", w.host, "done", "console") }), |
| 57 | "another host's entry": with(func(e *MoveLogEntry) { e.Body = entry("move", "elsewhere.example", "done", "console") }), |
| 58 | "a name that is not the identity": with(func(e *MoveLogEntry) { e.Name = "opened" }), |
| 59 | "a path": with(func(e *MoveLogEntry) { e.Name = "../source" }), |
| 60 | "a move id that is a path": with(func(e *MoveLogEntry) { e.Op = "../move" }), |
| 61 | "a host with no claim": with(func(e *MoveLogEntry) { |
| 62 | e.Host = "gone.example" |
| 63 | e.Body = entry("move", "gone.example", "done", "console") |
| 64 | }), |
| 65 | "too big": with(func(e *MoveLogEntry) { e.Body = bigBody }), |
| 66 | } { |
| 67 | _, got[what] = lw.WriteMoveLog(ctx, e) |
| 68 | } |
| 69 | return nil |
| 70 | }) |
| 71 | }) |
| 72 | if err != nil { |
| 73 | t.Fatal(err) |
| 74 | } |
| 75 | if !created || again { |
| 76 | t.Fatalf("created=%v, a replay of the same bytes created=%v", created, again) |
| 77 | } |
| 78 | for what, err := range got { |
| 79 | want := ErrIdentity |
| 80 | if what == "other bytes" { |
| 81 | want = ErrMoveLogTaken |
| 82 | } |
| 83 | if !errors.Is(err, want) { |
| 84 | t.Errorf("%s: %v, want %v", what, err, want) |
| 85 | } |
| 86 | } |
| 87 | prefix := moves.ReportPrefix("bab", "app", w.host, "move") |
| 88 | raw, _, err := w.raw.Get(ctx, moves.LogKeyFor(prefix, "done")) |
| 89 | if err != nil || string(raw) != string(good.Body) { |
| 90 | t.Fatalf("the object is not the entry's bytes: %s %v", raw, err) |
| 91 | } |
| 92 | if logged, err := moves.ReadLog(ctx, w.raw, prefix); err != nil || len(logged) != 1 { |
| 93 | t.Fatalf("log %+v %v", logged, err) |
| 94 | } |
| 95 | if _, err := saved.WriteMoveLog(ctx, with(func(e *MoveLogEntry) { e.Name = "late"; e.Body = entry("move", w.host, "late", "console") })); !errors.Is(err, ErrPending) { |
| 96 | t.Fatalf("a writer kept past its hook wrote: %v", err) |
| 97 | } |
| 98 | if err := runMoveLogDrain(ctx, func(context.Context, *MoveLogWriter) error { return nil }); !errors.Is(err, ErrContract) { |
| 99 | t.Fatalf("a writer minted outside a recovery pass: %v", err) |
| 100 | } |
| 101 | } |
| 102 | |
| 103 | // A closed move is named by its record's last move alone, so the claim |
| 104 | // holder's incarnation is what refuses a stale row there: the instance |
| 105 | // recreated under the same names, whose record may still close the same |
| 106 | // move id. |
| 107 | func TestMoveLogWriterRefusesAStaleIncarnationOfAClosedMove(t *testing.T) { |
| 108 | w := newC7World(t, func(c *hostclaim.Claim, r *provision.Record, _ *AppAdmission) { |
| 109 | r.LastMove = &provision.MoveOutcome{ID: r.Move.ID, Outcome: "done", At: r.CreatedAt} |
| 110 | r.Move = nil |
| 111 | }) |
| 112 | ctx := context.Background() |
| 113 | body, err := json.Marshal(moves.LogEvent{V: 1, Op: "move", Host: w.host, Event: "done", Identity: "done", Actor: "console", At: w.now}) |
| 114 | if err != nil { |
| 115 | t.Fatal(err) |
| 116 | } |
| 117 | var stale, current error |
| 118 | err = w.control.RunRecovery(ctx, func(ctx context.Context) error { |
| 119 | return runMoveLogDrain(ctx, func(ctx context.Context, lw *MoveLogWriter) error { |
| 120 | _, stale = lw.WriteMoveLog(ctx, MoveLogEntry{Host: w.host, Op: "move", Incarnation: strings.Repeat("ef", 32), Name: "done", Body: body}) |
| 121 | _, current = lw.WriteMoveLog(ctx, MoveLogEntry{Host: w.host, Op: "move", Incarnation: w.inc, Name: "done", Body: body}) |
| 122 | return nil |
| 123 | }) |
| 124 | }) |
| 125 | if err != nil { |
| 126 | t.Fatal(err) |
| 127 | } |
| 128 | if !errors.Is(stale, ErrIdentity) || current != nil { |
| 129 | t.Fatalf("stale incarnation: %v (want refused); current: %v", stale, current) |
| 130 | } |
| 131 | } |
| 132 | |
amadan