| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/ed25519" |
| 6 | "encoding/binary" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "net/http" |
| 12 | "net/http/httptest" |
| 13 | "os" |
| 14 | "path/filepath" |
| 15 | "strings" |
| 16 | "syscall" |
| 17 | "testing" |
| 18 | "time" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/appconfig" |
| 21 | "github.com/carlosframework/platform/internal/hostclaim" |
| 22 | "github.com/carlosframework/platform/internal/lease" |
| 23 | "github.com/carlosframework/platform/internal/moves" |
| 24 | "github.com/carlosframework/platform/internal/provision" |
| 25 | "github.com/carlosframework/platform/internal/registry" |
| 26 | "github.com/carlosframework/platform/internal/release" |
| 27 | ) |
| 28 | |
| 29 | // failing names the checks that did not pass, with their reasons. |
| 30 | func failing(r *moves.TargetReport) map[string]string { |
| 31 | out := map[string]string{} |
| 32 | for _, id := range moves.AllChecks { |
| 33 | if c := r.Checks[id]; !c.OK { |
| 34 | out[id] = c.Reason |
| 35 | } |
| 36 | } |
| 37 | return out |
| 38 | } |
| 39 | |
| 40 | // The target's report flags each check on its own (§7.3): from a staged |
| 41 | // move.json, a serving-then-moving claim that has not touched uk, and no |
| 42 | // archive, T9 fails on touched and T5 on the archive while T1 passes; once |
| 43 | // the claim touches uk and the source stages an archive, both pass. Each |
| 44 | // injectable hook, failing, fails its own check alone. |
| 45 | func TestTargetReportPassFlagsEachCheck(t *testing.T) { |
| 46 | t.Parallel() |
| 47 | ctx := context.Background() |
| 48 | h := newPhase1Harness(t) |
| 49 | h.start(t, "manual") |
| 50 | body, _ := json.Marshal(moves.MoveJSON{MoveID: p1ID, Incarnation: h.inc, Host: p1Host, From: "ie", To: "uk"}) |
| 51 | h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "m", Size: int64(len(body))}, string(body)}) |
| 52 | tc := h.targetCfg() |
| 53 | h.targetPass(t, tc) |
| 54 | r := h.targetReport(t) |
| 55 | if r == nil { |
| 56 | t.Fatal("no target report") |
| 57 | } |
| 58 | f := failing(r) |
| 59 | if !strings.Contains(f[moves.CheckClaim], "touched") || f[moves.CheckClaim] != t9Untouched("uk") { |
| 60 | t.Fatalf("T9 = %q, want the touched sentence for uk (%v)", f[moves.CheckClaim], f) |
| 61 | } |
| 62 | if !strings.Contains(f[moves.CheckArchive], "no archive") { |
| 63 | t.Fatalf("T5 = %q, want the no-archive sentence", f[moves.CheckArchive]) |
| 64 | } |
| 65 | if _, bad := f[moves.CheckProtocol]; bad { |
| 66 | t.Fatalf("T1 failed: %v", f) |
| 67 | } |
| 68 | if r.Box != "uk-box" || r.Region != "uk" || r.MoveID != p1ID { |
| 69 | t.Fatalf("report identity %+v", r) |
| 70 | } |
| 71 | |
| 72 | // Touch uk and uk-box, then let the source stage its archive. |
| 73 | obs, _ := hostclaim.Read(ctx, h.s, p1Host) |
| 74 | if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.Touch(p1ID, []string{"uk"}, []string{"uk-box"})); err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | h.pass(t) |
| 78 | h.targetPass(t, tc) |
| 79 | if f := failing(h.targetReport(t)); len(f) != 0 { |
| 80 | t.Fatalf("after touch and archive: %v", f) |
| 81 | } |
| 82 | |
| 83 | // Each hook alone. |
| 84 | hookErr := errors.New("injected") |
| 85 | for name, c := range map[string]struct { |
| 86 | check string |
| 87 | edit func(*moveCfg) |
| 88 | want string |
| 89 | }{ |
| 90 | "ComposeHeldEnv": {moves.CheckEnv, func(c *moveCfg) { |
| 91 | c.ComposeHeldEnv = func(context.Context, release.Store, provision.Record, []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) { |
| 92 | return nil, nil, nil, provision.MoveInputs{}, hookErr |
| 93 | } |
| 94 | }, t3NoCompose}, |
| 95 | "ComposeHeldEnv differs": {moves.CheckEnv, func(c *moveCfg) { |
| 96 | inner := c.ComposeHeldEnv |
| 97 | c.ComposeHeldEnv = func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) { |
| 98 | env, inst, sh, in, err := inner(ctx, s, rec, carried) |
| 99 | env["APP_MODE"] = "staging" |
| 100 | return env, inst, sh, in, err |
| 101 | } |
| 102 | }, t3Differs([]string{"APP_MODE (differs)"})}, |
| 103 | "ProbeStore": {moves.CheckStorage, func(c *moveCfg) { |
| 104 | c.ProbeStore = func(context.Context, map[string]string, string) (string, error) { return "MEDIA", hookErr } |
| 105 | }, t4Failed("MEDIA")}, |
| 106 | "DiskFree": {moves.CheckDisk, func(c *moveCfg) { |
| 107 | c.DiskFree = func(string) (uint64, uint64, error) { return 0, 0, hookErr } |
| 108 | }, t6Unmeasured}, |
| 109 | "ReleaseReady": {moves.CheckRelease, func(c *moveCfg) { |
| 110 | c.ReleaseReady = func(context.Context, release.Store, provision.Record, string) moves.Check { |
| 111 | return moves.Check{Reason: t7Waiting} |
| 112 | } |
| 113 | }, t7Waiting}, |
| 114 | "LaneProbe": {moves.CheckLane, func(c *moveCfg) { |
| 115 | c.LaneProbe = func(context.Context, string, string, string) error { return hookErr } |
| 116 | }, t10Failed}, |
| 117 | } { |
| 118 | t.Run(name, func(t *testing.T) { |
| 119 | hc := tc |
| 120 | c.edit(&hc) |
| 121 | h.targetPass(t, hc) |
| 122 | f := failing(h.targetReport(t)) |
| 123 | // T4 probes the stores bound in T3's composition, so without one |
| 124 | // it fails closed with T3's reason: the one dependency. |
| 125 | if name == "ComposeHeldEnv" && f[moves.CheckStorage] == t3NoCompose { |
| 126 | delete(f, moves.CheckStorage) |
| 127 | } |
| 128 | if len(f) != 1 || f[c.check] != c.want { |
| 129 | t.Fatalf("%s failing: %v, want only %s = %q", name, f, c.check, c.want) |
| 130 | } |
| 131 | }) |
| 132 | } |
| 133 | |
| 134 | // T9 owns move.json's validation: one naming another move fails it |
| 135 | // with the lane sentence. |
| 136 | other, _ := json.Marshal(moves.MoveJSON{MoveID: p1ID2, Incarnation: h.inc, Host: p1Host, From: "ie", To: "uk"}) |
| 137 | h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "o", Size: int64(len(other))}, string(other)}) |
| 138 | h.targetPass(t, tc) |
| 139 | if f := failing(h.targetReport(t)); f[moves.CheckClaim] != t9OtherMove { |
| 140 | t.Fatalf("T9 with another move's move.json: %v", f) |
| 141 | } |
| 142 | wrongTo, _ := json.Marshal(moves.MoveJSON{MoveID: p1ID, Incarnation: h.inc, Host: p1Host, From: "ie", To: "se"}) |
| 143 | h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "w", Size: int64(len(wrongTo))}, string(wrongTo)}) |
| 144 | h.targetPass(t, tc) |
| 145 | if f := failing(h.targetReport(t)); f[moves.CheckClaim] != t9OtherMove { |
| 146 | t.Fatalf("T9 with a move.json for another region: %v", f) |
| 147 | } |
| 148 | h.f.store("data-uk", moves.MoveJSONKey(p1Host, p1ID), fakeMoveObject{lease.MirrorObject{ETag: "m", Size: int64(len(body))}, string(body)}) |
| 149 | |
| 150 | // T5 checks the archive against its manifest: a manifest whose entry |
| 151 | // digest was changed under the same version is refused. |
| 152 | sr := h.sourceReport(t) |
| 153 | mkey := sr.Archive.ManifestKey |
| 154 | var man map[string]any |
| 155 | if err := json.Unmarshal([]byte(h.f.buckets["data-uk"][mkey].body), &man); err != nil { |
| 156 | t.Fatal(err) |
| 157 | } |
| 158 | entries := man["entries"].([]any) |
| 159 | entries[0].(map[string]any)["sha256"] = "sha256:" + strings.Repeat("0", 64) |
| 160 | tampered, _ := json.Marshal(man) |
| 161 | good := h.f.buckets["data-uk"][mkey] |
| 162 | h.f.store("data-uk", mkey, fakeMoveObject{good.MirrorObject, string(tampered)}) |
| 163 | // A version is immutable in S3, so T5 keeps its verdict per version; |
| 164 | // the fake rewrote one in place, so the target starts with no verdicts. |
| 165 | h.worker2.archives = nil |
| 166 | h.targetPass(t, tc) |
| 167 | if f := failing(h.targetReport(t)); f[moves.CheckArchive] != t5Refused { |
| 168 | t.Fatalf("T5 with a tampered manifest: %v", f) |
| 169 | } |
| 170 | h.f.store("data-uk", mkey, good) |
| 171 | h.worker2.archives = nil |
| 172 | |
| 173 | // Only the named target writes a report. |
| 174 | h3 := newPhase1Harness(t) |
| 175 | h3.start(t, "manual") |
| 176 | notTarget := h3.targetCfg() |
| 177 | notTarget.Box = "uk-box-2" |
| 178 | h3.targetPass(t, notTarget) |
| 179 | if r := h3.targetReport(t); r != nil { |
| 180 | t.Fatalf("a box that is not the move's target wrote a report: %+v", r) |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | // T3 composes the flat held row's environment on a scratch directory |
| 185 | // (ruling F19): the rendered bundle and the carried .env, through |
| 186 | // LaunchEnvironment. A carried variable overriding a rendered one is |
| 187 | // reported as shadowed (round-6 finding 7), with no instance overlay. |
| 188 | func TestComposeHeldEnvReportsShadowedNames(t *testing.T) { |
| 189 | ctx := context.Background() |
| 190 | s := release.NewDirStore(t.TempDir()) |
| 191 | privB64, _, err := appconfig.GenerateKey() |
| 192 | if err != nil { |
| 193 | t.Fatal(err) |
| 194 | } |
| 195 | priv, _ := appconfig.ParsePrivateKey(privB64, "test") |
| 196 | body, _ := json.Marshal(appconfig.Manifest{Schema: 1, App: "app", Vars: map[string]appconfig.Var{ |
| 197 | "APP_MODE": {Kind: "env", Value: "rendered"}, "ONLY_RENDERED": {Kind: "env", Value: "1"}}}) |
| 198 | etag, err := s.Put(ctx, appconfig.Key("bab", "app"), body, "") |
| 199 | if err != nil { |
| 200 | t.Fatal(err) |
| 201 | } |
| 202 | rec := provision.Record{Host: p1Host, Account: "bab", App: "app"} |
| 203 | env, _, shadowed, in, err := composeHeldEnv(priv, "uk", true)(ctx, s, rec, []byte("APP_MODE=carried\nCARRIED=1\n")) |
| 204 | if err != nil { |
| 205 | t.Fatal(err) |
| 206 | } |
| 207 | if strings.Join(shadowed, ",") != "APP_MODE" { |
| 208 | t.Fatalf("shadowed = %v", shadowed) |
| 209 | } |
| 210 | if env["APP_MODE"] != "carried" || env["ONLY_RENDERED"] != "1" || env["CARRIED"] != "1" || env["CARLOS_REGION"] != "uk" || |
| 211 | env["CARLOS_PURGE"] != "header" || env["CARLOS_CONTROL_SOCKET"] != provision.DefaultSocket(p1Host)+".control" { |
| 212 | t.Fatalf("composed env lacks an expected variable (names: %v)", envNames(env)) |
| 213 | } |
| 214 | if in.BundleETags["default"] != etag || in.OverlayETag != "" { |
| 215 | t.Fatalf("inputs %+v, want the bundle etag %q", in, etag) |
| 216 | } |
| 217 | // No cache on the live box: no CARLOS_PURGE. |
| 218 | env, _, _, _, err = composeHeldEnv(priv, "uk", false)(ctx, s, rec, nil) |
| 219 | if err != nil { |
| 220 | t.Fatal(err) |
| 221 | } |
| 222 | if _, ok := env["CARLOS_PURGE"]; ok { |
| 223 | t.Fatal("CARLOS_PURGE composed for a box with no cache") |
| 224 | } |
| 225 | |
| 226 | // Through the report: the shadowed name is a note, never a failure. |
| 227 | h := newPhase1Harness(t) |
| 228 | h.start(t, "manual") |
| 229 | h.pass(t) |
| 230 | tc := h.targetCfg() |
| 231 | inner := tc.ComposeHeldEnv |
| 232 | tc.ComposeHeldEnv = func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) { |
| 233 | env, inst, _, in, err := inner(ctx, s, rec, carried) |
| 234 | return env, inst, []string{"APP_MODE"}, in, err |
| 235 | } |
| 236 | h.targetPass(t, tc) |
| 237 | r := h.targetReport(t) |
| 238 | if !r.Checks[moves.CheckEnv].OK || len(r.Notes) != 1 || r.Notes[0] != t3Shadowed("APP_MODE") { |
| 239 | t.Fatalf("T3 %+v notes %v", r.Checks[moves.CheckEnv], r.Notes) |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | func envNames(env map[string]string) []string { |
| 244 | var out []string |
| 245 | for k := range env { |
| 246 | out = append(out, k) |
| 247 | } |
| 248 | return out |
| 249 | } |
| 250 | |
| 251 | // Round-6 finding 3: on a hardened box the agent cannot write the version |
| 252 | // store. From a MoveStart'ed record with no target registry row, T7 leaves |
| 253 | // a held-stage request naming the source report's Release and touches the |
| 254 | // stage-requested marker; the root pass (ServeHeldStageRequests under |
| 255 | // heldStageAuthority) materializes it into optDir; the next pass finds |
| 256 | // LiveVersion equal and T7 passes. A request naming another version, or an |
| 257 | // op the claim does not name, is refused and live/<host> is untouched. |
| 258 | func TestPreparingRelaysHeldStageOnAHardenedBox(t *testing.T) { |
| 259 | ctx := context.Background() |
| 260 | h := newPhase1Harness(t) |
| 261 | h.start(t, "manual") |
| 262 | shipPinnedV7(t, h.s) |
| 263 | h.pass(t) // the source report names Release v7 |
| 264 | |
| 265 | reg := testRegistry(t) |
| 266 | targetData, targetOpt := t.TempDir(), t.TempDir() |
| 267 | prev := heldMaterialize |
| 268 | t.Cleanup(func() { heldMaterialize = prev }) |
| 269 | heldMaterialize = func(context.Context, release.Store, ed25519.PublicKey, *release.DigestRatchet, string, registry.Route, string, *release.ProofGate, func(func() error) error) (*release.Manifest, error) { |
| 270 | return nil, fmt.Errorf("%w: %w", release.ErrStageBlocked, &os.PathError{Op: "mkdir", Path: "/opt/carlos/versions", Err: syscall.EROFS}) |
| 271 | } |
| 272 | tc := h.targetCfg() |
| 273 | tc.DataDir = targetData |
| 274 | tc.ReleaseReady = heldReleaseReady(reg, targetOpt, targetData, "uk-box", hsPub, hsRatchet(t), offProofGate(t), stageRequester(targetData)) |
| 275 | h.targetPass(t, tc) |
| 276 | if f := failing(h.targetReport(t)); f[moves.CheckRelease] != t7Waiting { |
| 277 | t.Fatalf("T7 = %q, want %q", f[moves.CheckRelease], t7Waiting) |
| 278 | } |
| 279 | raw, err := os.ReadFile(filepath.Join(release.HeldStageDir(targetData), p1Host+".json")) |
| 280 | if err != nil { |
| 281 | t.Fatal(err) |
| 282 | } |
| 283 | var req release.HeldStageRequest |
| 284 | if err := json.Unmarshal(raw, &req); err != nil || req.Version != "v7" || req.Op != p1ID || req.Box != "uk-box" || req.Account != "bab" || req.App != "app" { |
| 285 | t.Fatalf("request %s (%v)", raw, err) |
| 286 | } |
| 287 | if !fileExistsForTest(filepath.Join(targetData, stageRequestMarker)) { |
| 288 | t.Fatal("the stage-requested marker was not touched") |
| 289 | } |
| 290 | |
| 291 | // The root pass, with the relay's own authority, materializes it. |
| 292 | heldMaterialize = prev |
| 293 | if err := release.ServeHeldStageRequestsWithPins(ctx, h.s, hsPub, hsRatchet(t), targetOpt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box", targetData)); err != nil { |
| 294 | t.Fatal(err) |
| 295 | } |
| 296 | if v, err := release.LiveVersion(targetOpt, p1Host); err != nil || v != "v7" { |
| 297 | t.Fatalf("LiveVersion = %q, %v", v, err) |
| 298 | } |
| 299 | h.targetPass(t, tc) |
| 300 | if f := failing(h.targetReport(t)); f[moves.CheckRelease] != "" { |
| 301 | t.Fatalf("T7 after the root stage: %q", f[moves.CheckRelease]) |
| 302 | } |
| 303 | |
| 304 | // Requests the authority refuses: another version, another op. |
| 305 | for _, bad := range []release.HeldStageRequest{ |
| 306 | {Account: "bab", App: "app", Host: p1Host, Version: "v8", Op: p1ID, Box: "uk-box"}, |
| 307 | {Account: "bab", App: "app", Host: p1Host, Version: "v7", Op: p1ID2, Box: "uk-box"}, |
| 308 | } { |
| 309 | opt := t.TempDir() |
| 310 | if err := writeHeldStageRequest(targetData, bad); err != nil { |
| 311 | t.Fatal(err) |
| 312 | } |
| 313 | if err := release.ServeHeldStageRequestsWithPins(ctx, h.s, hsPub, hsRatchet(t), opt, targetData, offProofGate(t), heldStageAuthority(reg, h.s, "uk-box", targetData)); err != nil { |
| 314 | t.Fatal(err) |
| 315 | } |
| 316 | if v, _ := release.LiveVersion(opt, p1Host); v != "" { |
| 317 | t.Fatalf("a refused request (%s, %s) repointed live to %q", bad.Version, bad.Op, v) |
| 318 | } |
| 319 | } |
| 320 | } |
| 321 | |
| 322 | // T4 finds the store bindings in the composed environment by objstore's |
| 323 | // variable names; a prefix without its credentials is not a binding. |
| 324 | func TestStorePrefixesFollowTheDeliveredNames(t *testing.T) { |
| 325 | env := map[string]string{ |
| 326 | "MEDIA_BUCKET": "b", "MEDIA_ACCESS_KEY": "k", "MEDIA_SECRET_KEY": "s", |
| 327 | "STORE_BUCKET": "b2", "STORE_ACCESS_KEY": "k2", "STORE_SECRET_KEY": "s2", "STORE_REGION": "eu-west-1", |
| 328 | "HALF_BUCKET": "b3", "HALF_ACCESS_KEY": "k3", |
| 329 | "_BUCKET": "x", |
| 330 | } |
| 331 | if got := strings.Join(storePrefixes(env), ","); got != "MEDIA,STORE" { |
| 332 | t.Fatalf("prefixes %q", got) |
| 333 | } |
| 334 | } |
| 335 | |
| 336 | // T10 in production: put-if-absent then delete by version; a probe object |
| 337 | // left behind is removed and the check fails for that pass only. |
| 338 | func TestLaneProbePutsAndDeletesByVersion(t *testing.T) { |
| 339 | ctx := context.Background() |
| 340 | f := newFakeMoveMirror() |
| 341 | vs := &fakeVersioned{f: f} |
| 342 | probe := laneProbe(func(context.Context, string, string) (lease.VersionedStore, error) { return vs, nil }) |
| 343 | if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err != nil { |
| 344 | t.Fatal(err) |
| 345 | } |
| 346 | if _, ok := f.buckets["data-ie"][moves.LaneKey(p1ID)]; ok { |
| 347 | t.Fatal("the probe object was left behind") |
| 348 | } |
| 349 | f.put("data-ie", moves.LaneKey(p1ID), "left") |
| 350 | f.setVersion("data-ie", moves.LaneKey(p1ID), "v-left") |
| 351 | if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err == nil { |
| 352 | t.Fatal("a left-behind probe object passed") |
| 353 | } |
| 354 | if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err != nil { |
| 355 | t.Fatalf("the next pass did not start clean: %v", err) |
| 356 | } |
| 357 | vs.deny = func(string, string, string) error { return errors.New("AccessDenied") } |
| 358 | if err := probe(ctx, "data-ie", "eu-west-1", p1ID); err == nil { |
| 359 | t.Fatal("a denied lane passed") |
| 360 | } |
| 361 | } |
| 362 | |
| 363 | // The disk measurement (O_DIRECT on linux, synced elsewhere) and the |
| 364 | // per-file sync rate both produce a usable, dated measurement. |
| 365 | func TestMeasureDiskIn(t *testing.T) { |
| 366 | t.Parallel() |
| 367 | if testing.Short() { |
| 368 | t.Skip("writes and reads 64 MiB") |
| 369 | } |
| 370 | dir := t.TempDir() |
| 371 | disk, files, err := measureDiskIn(dir, time.Now)(context.Background()) |
| 372 | if err != nil { |
| 373 | t.Fatal(err) |
| 374 | } |
| 375 | if disk.MiBPerS <= 0 || disk.At.IsZero() || disk.Bytes != 2*measureDiskSize { |
| 376 | t.Fatalf("disk %+v", disk) |
| 377 | } |
| 378 | if files.PerS <= 0 || files.At.IsZero() || files.Files != measureFsyncFiles { |
| 379 | t.Fatalf("fsync %+v", files) |
| 380 | } |
| 381 | if ents, _ := os.ReadDir(filepath.Join(dir, ".moves")); len(ents) != 0 { |
| 382 | t.Fatalf("the measurement left %d entries behind", len(ents)) |
| 383 | } |
| 384 | } |
| 385 | |
| 386 | // The regional S3 measurement writes the probe key outside moves/ with |
| 387 | // this box's own client and deletes it (F11). |
| 388 | func TestMeasureS3UsesTheProbeKeyAndCleansUp(t *testing.T) { |
| 389 | t.Parallel() |
| 390 | f := newFakeMoveMirror() |
| 391 | m := measureS3(func(context.Context) (lease.MirrorStore, error) { return f, nil }, "data-uk", (&stepClock{now: time.Now()}).Now) |
| 392 | r, err := m(context.Background(), moves.ProbeKey(p1Host, p1ID)) |
| 393 | if err != nil { |
| 394 | t.Fatal(err) |
| 395 | } |
| 396 | if r.MiBPerS <= 0 || r.At.IsZero() { |
| 397 | t.Fatalf("%+v", r) |
| 398 | } |
| 399 | if len(f.buckets["data-uk"]) != 0 { |
| 400 | t.Fatalf("the probe object was left behind: %v", f.buckets["data-uk"]) |
| 401 | } |
| 402 | if strings.Join(f.ops, ",") != "put data-uk/"+moves.ProbeKey(p1Host, p1ID)+",delete data-uk/"+moves.ProbeKey(p1Host, p1ID) { |
| 403 | t.Fatalf("ops %v", f.ops) |
| 404 | } |
| 405 | } |
| 406 | |
| 407 | // T7 on a box that can write its version store: the release is staged and |
| 408 | // published directly, under the same authority the root relay uses, so a |
| 409 | // claim that names another op publishes nothing. |
| 410 | func TestHeldReleaseReadyOnAWritableBox(t *testing.T) { |
| 411 | ctx := context.Background() |
| 412 | h := newPhase1Harness(t) |
| 413 | rec := h.start(t, "manual") |
| 414 | shipPinnedV7(t, h.s) |
| 415 | h.pass(t) |
| 416 | reg := testRegistry(t) |
| 417 | data, opt := t.TempDir(), t.TempDir() |
| 418 | ready := heldReleaseReady(reg, opt, data, "uk-box", hsPub, hsRatchet(t), offProofGate(t), nil) |
| 419 | if c := ready(ctx, h.s, rec, "v7"); !c.OK { |
| 420 | t.Fatalf("T7 = %+v", c) |
| 421 | } |
| 422 | if v, _ := release.LiveVersion(opt, p1Host); v != "v7" { |
| 423 | t.Fatalf("live = %q", v) |
| 424 | } |
| 425 | // The claim moves to another op while the release stages: the |
| 426 | // publication's authority refuses, nothing is published. |
| 427 | prev := heldMaterialize |
| 428 | t.Cleanup(func() { heldMaterialize = prev }) |
| 429 | heldMaterialize = func(ctx context.Context, s release.Store, pub ed25519.PublicKey, r *release.DigestRatchet, optDir string, rt registry.Route, v string, proof *release.ProofGate, guard func(func() error) error) (*release.Manifest, error) { |
| 430 | rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.ID = p1ID2 }) |
| 431 | return prev(ctx, s, pub, r, optDir, rt, v, proof, guard) |
| 432 | } |
| 433 | opt2 := t.TempDir() |
| 434 | if c := heldReleaseReady(reg, opt2, data, "uk-box", hsPub, hsRatchet(t), offProofGate(t), nil)(ctx, h.s, rec, "v7"); c.OK || c.Reason != t7Failed { |
| 435 | t.Fatalf("T7 when the op changed during staging = %+v", c) |
| 436 | } |
| 437 | if v, _ := release.LiveVersion(opt2, p1Host); v != "" { |
| 438 | t.Fatalf("published after the op changed: %q", v) |
| 439 | } |
| 440 | // Under another op from the start: refused before anything is staged. |
| 441 | heldMaterialize = prev |
| 442 | opt3 := t.TempDir() |
| 443 | if c := heldReleaseReady(reg, opt3, data, "uk-box", hsPub, hsRatchet(t), offProofGate(t), nil)(ctx, h.s, rec, "v7"); c.OK || c.Reason != t7Failed { |
| 444 | t.Fatalf("T7 under another op = %+v", c) |
| 445 | } |
| 446 | if _, err := os.Stat(filepath.Join(opt3, "versions")); !errors.Is(err, os.ErrNotExist) { |
| 447 | t.Fatalf("a release was staged for a stale op: %v", err) |
| 448 | } |
| 449 | } |
| 450 | |
| 451 | // probeStore's SDK path, locally: against a minimal S3 endpoint it lists one |
| 452 | // key, then puts, gets and deletes .carlos-move-probe/<id>, in that order, |
| 453 | // path-style at the delivered endpoint, and a refusal fails it. (The real |
| 454 | // server is TestT4StorageProbeMinio's, in CI.) |
| 455 | func TestProbeStoreSpeaksS3(t *testing.T) { |
| 456 | var seen []string |
| 457 | objects := map[string][]byte{} |
| 458 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 459 | if !strings.Contains(r.Header.Get("Authorization"), "Credential=good/") { |
| 460 | w.WriteHeader(http.StatusForbidden) |
| 461 | io.WriteString(w, `<?xml version="1.0"?><Error><Code>AccessDenied</Code><Message>no</Message></Error>`) |
| 462 | return |
| 463 | } |
| 464 | key := strings.TrimPrefix(r.URL.Path, "/bkt") |
| 465 | switch { |
| 466 | case r.Method == http.MethodGet && r.URL.Query().Get("list-type") == "2": |
| 467 | seen = append(seen, "list") |
| 468 | io.WriteString(w, `<?xml version="1.0"?><ListBucketResult><Name>bkt</Name><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`) |
| 469 | case r.Method == http.MethodPut: |
| 470 | seen = append(seen, "put "+key) |
| 471 | b, _ := io.ReadAll(r.Body) |
| 472 | objects[key] = b |
| 473 | case r.Method == http.MethodGet: |
| 474 | seen = append(seen, "get "+key) |
| 475 | w.Write(objects[key]) |
| 476 | case r.Method == http.MethodDelete: |
| 477 | seen = append(seen, "delete "+key) |
| 478 | delete(objects, key) |
| 479 | w.WriteHeader(http.StatusNoContent) |
| 480 | default: |
| 481 | w.WriteHeader(http.StatusMethodNotAllowed) |
| 482 | } |
| 483 | })) |
| 484 | t.Cleanup(srv.Close) |
| 485 | env := map[string]string{"STORE_BUCKET": "bkt", "STORE_ENDPOINT": srv.URL, "STORE_ACCESS_KEY": "good", "STORE_SECRET_KEY": "s"} |
| 486 | if prefix, err := probeStores(context.Background(), env, p1ID); err != nil { |
| 487 | t.Fatalf("%s: %v", prefix, err) |
| 488 | } |
| 489 | k := "/.carlos-move-probe/" + p1ID |
| 490 | if got, want := strings.Join(seen, ","), "list,put "+k+",get "+k+",delete "+k; got != want { |
| 491 | t.Fatalf("requests %q, want %q", got, want) |
| 492 | } |
| 493 | if len(objects) != 0 { |
| 494 | t.Fatal("the probe object was left behind") |
| 495 | } |
| 496 | env["STORE_ACCESS_KEY"] = "bad" |
| 497 | if prefix, err := probeStores(context.Background(), env, p1ID); err == nil || prefix != "STORE" { |
| 498 | t.Fatalf("wrong credentials: %q %v", prefix, err) |
| 499 | } |
| 500 | } |
| 501 | |
| 502 | // T11 prices the database's logical size, read from the newest staged LTX |
| 503 | // header (Commit pages of PageSize), never the compressed bytes staged |
| 504 | // (Task 14 review finding 1). The source's local database file counts when |
| 505 | // it is larger; a newest file that is not LTX fails T6 and T11 closed. |
| 506 | func TestT11PricesTheLogicalDatabaseSize(t *testing.T) { |
| 507 | t.Parallel() |
| 508 | h := newPhase1Harness(t) |
| 509 | h.start(t, "manual") |
| 510 | newest := ltxForTest(t, 4096, 1<<18, 3, 3) // a 1 GiB database, one page in this file |
| 511 | if len(newest) > 64<<10 { |
| 512 | t.Fatalf("the fixture is %d bytes: not compressed enough to prove anything", len(newest)) |
| 513 | } |
| 514 | key := "instances/" + p1Host + "/" + h.inc + "/g-cur/0000/0000000000000003-0000000000000003.ltx" |
| 515 | h.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "newest", Size: int64(len(newest))}, string(newest)}) |
| 516 | h.pass(t) |
| 517 | h.targetPass(t, h.targetCfg()) |
| 518 | r := h.targetReport(t) |
| 519 | if r.DBMiB != 1024 { |
| 520 | t.Fatalf("T11 priced %v MiB, want the 1024 MiB the newest header names (staged bytes are about %d KiB)", r.DBMiB, len(newest)>>10) |
| 521 | } |
| 522 | |
| 523 | // A larger local database file on the source wins. |
| 524 | db, err := os.Create(provision.DefaultDBPath(h.cfg.DataDir, p1Host)) |
| 525 | if err != nil { |
| 526 | t.Fatal(err) |
| 527 | } |
| 528 | if err := db.Truncate(2 << 30); err != nil { // sparse |
| 529 | t.Fatal(err) |
| 530 | } |
| 531 | db.Close() |
| 532 | h.pass(t) |
| 533 | if sr := h.sourceReport(t); sr.DBBytes != 2<<30 { |
| 534 | t.Fatalf("source DBBytes %d", sr.DBBytes) |
| 535 | } |
| 536 | h.targetPass(t, h.targetCfg()) |
| 537 | if r := h.targetReport(t); r.DBMiB != 2048 { |
| 538 | t.Fatalf("T11 priced %v MiB with a 2 GiB source file", r.DBMiB) |
| 539 | } |
| 540 | |
| 541 | // The newest staged file is not LTX: the size is unknown, so T6 and |
| 542 | // T11 fail closed. |
| 543 | h2 := newPhase1Harness(t) |
| 544 | h2.start(t, "manual") |
| 545 | h2.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "junk", Size: 4}, "junk"}) |
| 546 | h2.pass(t) |
| 547 | h2.targetPass(t, h2.targetCfg()) |
| 548 | f := failing(h2.targetReport(t)) |
| 549 | if f[moves.CheckBudget] != t11Unmeasured || f[moves.CheckDisk] != t11Unmeasured { |
| 550 | t.Fatalf("an unreadable newest header: %v", f) |
| 551 | } |
| 552 | |
| 553 | // A header that parses but names no pages is no measurement either. |
| 554 | zero := append([]byte(nil), newest...) |
| 555 | binary.BigEndian.PutUint32(zero[12:], 0) // Commit |
| 556 | h3 := newPhase1Harness(t) |
| 557 | h3.start(t, "manual") |
| 558 | h3.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "zero", Size: int64(len(zero))}, string(zero)}) |
| 559 | h3.pass(t) |
| 560 | h3.targetPass(t, h3.targetCfg()) |
| 561 | if f := failing(h3.targetReport(t)); f[moves.CheckBudget] != t11Unmeasured { |
| 562 | t.Fatalf("a header naming zero pages: %v", f) |
| 563 | } |
| 564 | } |
| 565 | |
| 566 | // T5's verdicts (with the carried .env) are kept only while their move is |
| 567 | // in flight into this box. |
| 568 | func TestArchiveVerdictsAreEvictedWhenTheMoveEnds(t *testing.T) { |
| 569 | ctx := context.Background() |
| 570 | h := newPhase1Harness(t) |
| 571 | h.start(t, "manual") |
| 572 | h.pass(t) |
| 573 | h.targetPass(t, h.targetCfg()) |
| 574 | if len(h.worker2.archives) != 1 { |
| 575 | t.Fatalf("%d verdicts cached", len(h.worker2.archives)) |
| 576 | } |
| 577 | g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc} |
| 578 | if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedAbort(g)); err != nil { |
| 579 | t.Fatal(err) |
| 580 | } |
| 581 | // The source's pass begins the abort; the record stays aborting while |
| 582 | // the target region parks the lane (Plan 5 Task 1). The move is over for |
| 583 | // the target's caches already. |
| 584 | h.pass(t) |
| 585 | if m := h.read(t, p1Host).Move; m == nil || m.State != provision.MoveAborting { |
| 586 | t.Fatalf("the record during the park wait: %+v", m) |
| 587 | } |
| 588 | h.targetPass(t, h.targetCfg()) |
| 589 | if len(h.worker2.archives) != 0 { |
| 590 | t.Fatalf("a verdict outlived its move: %v", h.worker2.archives) |
| 591 | } |
| 592 | } |
| 593 | |
amadan