| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "strings" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/moves" |
| 13 | "github.com/carlosframework/platform/internal/provision" |
| 14 | "github.com/carlosframework/platform/internal/release" |
| 15 | ) |
| 16 | |
| 17 | // Ruling R-SPEC-43A-2 (Astra round 1 finding 3): before a cut seals the build |
| 18 | // state, the source box's consumed request record is merged into it under |
| 19 | // the lease (release.CarryRequest), on every lane: the legacy cut, the Phase |
| 20 | // 1 cut's final round and a dirty return's lane. |
| 21 | |
| 22 | // carryCall is one CarryRequest call, host bucket/prefix. |
| 23 | type carryCall struct{ host, bucket, prefix string } |
| 24 | |
| 25 | // The legacy cut carries the request before its byte-final round, after the |
| 26 | // final checkpoint, at the hibernate prefix the round copies; the carried |
| 27 | // state is what lands in staging. |
| 28 | func TestTheLegacyCutCarriesTheRequestBeforeItsFinalRound(t *testing.T) { |
| 29 | h := newMoveHarness(t) |
| 30 | host := "a.go.example.com" |
| 31 | h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting()) |
| 32 | h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| 33 | h.f.put("data-ie", "instances/"+host+"/build-state.json", "before") |
| 34 | var calls []carryCall |
| 35 | h.cfg.CarryRequest = func(_ context.Context, host, bucket, prefix string) error { |
| 36 | calls = append(calls, carryCall{host, bucket, prefix}) |
| 37 | h.f.ops = append(h.f.ops, "carry") |
| 38 | h.f.put(bucket, prefix+"/build-state.json", "carried") |
| 39 | return nil |
| 40 | } |
| 41 | if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| 42 | t.Fatal(err) |
| 43 | } |
| 44 | if len(calls) != 1 || calls[0] != (carryCall{host, "data-ie", "instances/" + host}) { |
| 45 | t.Fatalf("carry calls %+v; want one at data-ie/instances/%s", calls, host) |
| 46 | } |
| 47 | ck, carry, cp := -1, -1, -1 |
| 48 | for i, op := range h.f.ops { |
| 49 | switch { |
| 50 | case strings.HasPrefix(op, "checkpoint ") && ck == -1: |
| 51 | ck = i |
| 52 | case op == "carry": |
| 53 | carry = i |
| 54 | case strings.HasPrefix(op, "copy ") && cp == -1: |
| 55 | cp = i |
| 56 | } |
| 57 | } |
| 58 | if !(ck < carry && carry < cp) { |
| 59 | t.Fatalf("ops %v: want checkpoint, then the carry, then the final round", h.f.ops) |
| 60 | } |
| 61 | if got := h.f.bodies("data-uk", "moves/"+host+"/")["build-state.json"]; got != "carried" { |
| 62 | t.Fatalf("staged build state %q, want the carried one", got) |
| 63 | } |
| 64 | if got := h.read(t, host); got.Move != nil { |
| 65 | t.Fatalf("the cut did not complete: %+v", got.Move) |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | // A carry that fails refuses the cut (the pass logs the round's error and |
| 70 | // goes on to other hosts): nothing is copied, the warm cache is not purged, |
| 71 | // and the move stays cutting for the next pass. |
| 72 | func TestAFailedCarryRefusesTheLegacyCut(t *testing.T) { |
| 73 | h := newMoveHarness(t) |
| 74 | host := "a.go.example.com" |
| 75 | h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting()) |
| 76 | h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| 77 | db := filepath.Join(h.cfg.DataDir, host+".db") |
| 78 | if err := writeFileForTest(db, "warm"); err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | carried := 0 |
| 82 | h.cfg.CarryRequest = func(context.Context, string, string, string) error { carried++; return errors.New("s3: 503") } |
| 83 | if err := movesPass(context.Background(), h.s, h.worker()); err != nil { |
| 84 | t.Fatal(err) |
| 85 | } |
| 86 | if carried != 1 { |
| 87 | t.Fatalf("carry asked %d times, want 1", carried) |
| 88 | } |
| 89 | if _, err := os.Stat(db); err != nil { |
| 90 | t.Fatalf("the warm cache was purged past a failed carry: %v", err) |
| 91 | } |
| 92 | for _, op := range h.f.ops { |
| 93 | if strings.HasPrefix(op, "copy ") { |
| 94 | t.Fatalf("ops %v: the final round ran past a failed carry", h.f.ops) |
| 95 | } |
| 96 | } |
| 97 | if got := h.read(t, host); got.Move == nil || got.Move.State != provision.MoveCutting { |
| 98 | t.Fatalf("move %+v, want still cutting", got.Move) |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | // The Phase 1 cut's final round carries the request at the claim holder's |
| 103 | // location before it copies the state for the seal. |
| 104 | func TestThePhase1CutCarriesTheRequestBeforeItsSeal(t *testing.T) { |
| 105 | h := newCutHarness(t) |
| 106 | storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6") |
| 107 | h.ready(t) |
| 108 | var calls []carryCall |
| 109 | h.cfg.CarryRequest = func(_ context.Context, host, bucket, prefix string) error { |
| 110 | calls = append(calls, carryCall{host, bucket, prefix}) |
| 111 | storeState(h.phase1Harness, bucket, prefix+"/build-state.json", "carried") |
| 112 | return nil |
| 113 | } |
| 114 | h.cut(t) |
| 115 | h.pass(t) |
| 116 | if m := h.move(t); m.State != provision.MoveLanding { |
| 117 | t.Fatalf("cut did not finish: %+v", m) |
| 118 | } |
| 119 | if len(calls) == 0 || calls[len(calls)-1] != (carryCall{p1Host, "data-ie", cutRoot}) { |
| 120 | t.Fatalf("carry calls %+v; want one at data-ie/%s", calls, cutRoot) |
| 121 | } |
| 122 | key := moves.BuildStateKey(p1Host, p1ID) |
| 123 | if got := h.f.buckets["data-uk"][key].body; got != p1StateBody("carried") { |
| 124 | t.Fatalf("the sealed lane holds %q, want the carried state", got) |
| 125 | } |
| 126 | if seal := h.seal(t); len(seal.State) != 1 || seal.State[0].VersionID != h.f.versionOf("data-uk", key) { |
| 127 | t.Fatalf("seal state %+v", seal.State) |
| 128 | } |
| 129 | } |
| 130 | |
| 131 | // A dirty return carries the target's request record into the state beside |
| 132 | // the target's lease before its lane copies it back. |
| 133 | func TestTheReturnLaneCarriesTheRequest(t *testing.T) { |
| 134 | t.Parallel() |
| 135 | h := &rollbackHarness{landingHarness: newLandingHarness(t)} |
| 136 | storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6") |
| 137 | h.flip(t) |
| 138 | h.land(t) |
| 139 | h.installed(t) |
| 140 | h.step(t, provision.LandRestored, provision.LandProbing) |
| 141 | h.tpos = h.n |
| 142 | storeState(h.phase1Harness, "data-uk", cutRoot+"/build-state.json", "v8") |
| 143 | h.uncover(t) |
| 144 | h.rollBack(t) |
| 145 | var calls []carryCall |
| 146 | h.carry = func(_ context.Context, host, bucket, prefix string) error { |
| 147 | calls = append(calls, carryCall{host, bucket, prefix}) |
| 148 | storeState(h.phase1Harness, bucket, prefix+"/build-state.json", "carried") |
| 149 | return nil |
| 150 | } |
| 151 | h.rollbackPass(t) |
| 152 | rb := h.rb(t) |
| 153 | if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned { |
| 154 | t.Fatalf("rollback %+v", rb) |
| 155 | } |
| 156 | if len(calls) == 0 || calls[len(calls)-1] != (carryCall{p1Host, "data-uk", cutRoot}) { |
| 157 | t.Fatalf("carry calls %+v; want one at data-uk/%s", calls, cutRoot) |
| 158 | } |
| 159 | if _, body := returnSealState(t, h, rb.ID); body != p1StateBody("carried") { |
| 160 | t.Fatalf("the return lane holds %q, want the carried state", body) |
| 161 | } |
| 162 | } |
| 163 | |
| 164 | // buildStates.carryRequest, the hook's production body: under the host's |
| 165 | // lock it merges root's consumed record into the object at the location |
| 166 | // given (release.CarryRequest), last_executed untouched, and writes only |
| 167 | // when something changes. |
| 168 | func TestCarryRequestWritesTheConsumedRecordAtTheLocation(t *testing.T) { |
| 169 | b, s, reg := testBuildStates(t) |
| 170 | ctx := context.Background() |
| 171 | if err := put(t, b, ctx, astraHost, release.BuildState{LastExecuted: "A"}); err != nil { |
| 172 | t.Fatal(err) |
| 173 | } |
| 174 | // The cut holds the row before it carries (ruling R-CARRY-FENCE). |
| 175 | if err := reg.SetStatus(astraHost, "moving"); err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | key := release.BuildStateKey("inst/" + astraHost) |
| 179 | etag := mustEtag(t, s, key) |
| 180 | // No consumed file: nothing to carry, nothing written. |
| 181 | if err := b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost); err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | if mustEtag(t, s, key) != etag { |
| 185 | t.Fatal("a carry with no consumed file rewrote the object") |
| 186 | } |
| 187 | writeConsumed(t, b, release.Consumed{Resolved: &release.ResolvedRequest{ID: 1, Target: "T"}, Completed: 1, CompletedFailed: true}) |
| 188 | if err := b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost); err != nil { |
| 189 | t.Fatal(err) |
| 190 | } |
| 191 | st, _, err := getState(ctx, s, key, astraHost) |
| 192 | if err != nil || st == nil || st.LastExecuted != "A" || st.Completed != 1 || !st.CompletedFailed { |
| 193 | t.Fatalf("state %+v, %v; want A with R1 completed failed", st, err) |
| 194 | } |
| 195 | etag = mustEtag(t, s, key) |
| 196 | if err := b.carryRequest(ctx, astraHost, "hib", "inst/"+astraHost); err != nil { |
| 197 | t.Fatal(err) |
| 198 | } |
| 199 | if mustEtag(t, s, key) != etag { |
| 200 | t.Fatal("a carry with nothing new rewrote the object") |
| 201 | } |
| 202 | // The location given wins over the hibernate prefix (a claim holder's). |
| 203 | other := release.NewDirStore(t.TempDir()) |
| 204 | b.storeFor = storesOf(map[string]release.Store{"hib": s, "data": other}) |
| 205 | if err := b.carryRequest(ctx, astraHost, "data", "instances/"+astraHost+"/inc1"); err != nil { |
| 206 | t.Fatal(err) |
| 207 | } |
| 208 | if st, _, _ := getState(ctx, other, release.BuildStateKey("instances/"+astraHost+"/inc1"), astraHost); st == nil || st.Completed != 1 { |
| 209 | t.Fatalf("state at the location = %+v", st) |
| 210 | } |
| 211 | } |
| 212 | |
amadan