| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "os/exec" |
| 10 | "path/filepath" |
| 11 | "strconv" |
| 12 | "strings" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/genselect" |
| 17 | "github.com/carlosframework/platform/internal/hostbox" |
| 18 | "github.com/carlosframework/platform/internal/hostclaim" |
| 19 | "github.com/carlosframework/platform/internal/hostfiles" |
| 20 | "github.com/carlosframework/platform/internal/hostwrite" |
| 21 | "github.com/carlosframework/platform/internal/lease/leasetest" |
| 22 | "github.com/carlosframework/platform/internal/moves" |
| 23 | "github.com/carlosframework/platform/internal/registry" |
| 24 | "github.com/carlosframework/platform/internal/release" |
| 25 | ) |
| 26 | |
| 27 | // relocationProbeRig is P's first probe on box-1 through the claim's own |
| 28 | // runtime (the recovery fixture's controller), the relocation owners bound |
| 29 | // as claimGroupDWiring binds them, the probe attempt counted and pending. |
| 30 | type relocationProbeRig struct { |
| 31 | f *recoveryFixture |
| 32 | bucket *leasetest.VersionedBucket |
| 33 | files *release.DirStore |
| 34 | c8 *roFiles |
| 35 | inputs *RelocationWakeInputs |
| 36 | c *recoveryController |
| 37 | h *hostclaim.Holder |
| 38 | o hostclaim.Observed |
| 39 | im moves.InstallManifest |
| 40 | auth hostclaim.PreparedAuthority |
| 41 | route registry.Route |
| 42 | } |
| 43 | |
| 44 | // roProbeLaunch is the serving launch source's result for the rig: the |
| 45 | // live link hashed as build v1, staged under v1. |
| 46 | func roProbeLaunch() ClaimServingLaunch { |
| 47 | return ClaimServingLaunch{Identity: hostwrite.Identity{Config: "sha256:config", Release: "sha256:release"}, Executable: "/usr/bin/true", ExecutableSHA256: strings.Repeat("a", 64), Version: "v1", |
| 48 | Staged: release.StagedIdentity{Version: "v1", StagedPath: "/opt/carlos/versions/v1/bab/jam", Digest: strings.Repeat("a", 64)}} |
| 49 | } |
| 50 | |
| 51 | func newRelocationProbeRig(t *testing.T, launch ClaimServingLaunch) *relocationProbeRig { |
| 52 | t.Helper() |
| 53 | ctx := t.Context() |
| 54 | f := newRecoveryFixture(t) |
| 55 | bucket := leasetest.NewVersionedBucket() |
| 56 | bucket.AddBucket("data") |
| 57 | files := release.NewDirStore(t.TempDir()) |
| 58 | o := f.read() |
| 59 | h := o.Claim.Holder |
| 60 | prefix := moves.GenerationPrefix(h.Location, roP) |
| 61 | snap, anchor := moves.SnapshotKey(prefix, 42), moves.AnchorKey(prefix, 42) |
| 62 | im := moves.InstallManifest{V: 1, Op: roOp, Host: recoveryHost, N: 42, SnapshotKey: snap, SnapshotVersion: bucket.Put("data", snap, ltxHeader(t, 1, 42)), SnapshotCRC64: 7, |
| 63 | AnchorKey: anchor, AnchorVersion: bucket.Put("data", anchor, ltxHeader(t, 42, 42)), IntegrityCheck: "ok", DBSHA256: strings.Repeat("d", 64)} |
| 64 | imRaw := mustJSON(t, im) |
| 65 | bucket.Put("data", moves.InstallManifestKey(prefix), imRaw) |
| 66 | pm := mustJSON(t, hostfiles.Manifest{V: hostfiles.ManifestVersion, Entries: map[string]hostfiles.Entry{}, Platform: map[string]hostfiles.Entry{}}) |
| 67 | if _, err := files.Put(ctx, hostfiles.ManifestKey(moves.GenerationFilesPrefix(h.Location, roP)), pm, ""); err != nil { |
| 68 | t.Fatal(err) |
| 69 | } |
| 70 | imSum, pmSum := sha256.Sum256(imRaw), sha256.Sum256(pm) |
| 71 | auth := hostclaim.PreparedAuthority{Generation: roP, InstallManifest: moves.InstallManifestKey(prefix), SHA256: hex.EncodeToString(imSum[:]), SnapshotTXID: 42, SourceFilesSeq: 5, FilesManifestSHA256: hex.EncodeToString(pmSum[:])} |
| 72 | ref := hostclaim.RecoveryPin{Key: "control/relocations/" + recoveryHost + "/" + roOp + "/pin-1.json", SHA256: strings.Repeat("e", 64), Generation: h.Generation, MaxTXID: 42, FilesSeq: 5, Release: "v1"} |
| 73 | o, err := hostclaim.Update(ctx, f.claims, o, func(c *hostclaim.Claim) error { |
| 74 | c.Holder.Stamp = c.Gen + 1 |
| 75 | auth.InstallStamp = c.Holder.Stamp |
| 76 | c.Op = hostclaim.Op{Kind: hostclaim.KindRelocating, ID: roOp, Phase: hostclaim.PhaseRelocatingProbing, Exec: "box-1", Participants: []string{"box-0", "box-1"}, |
| 77 | Intent: map[string]string{"region": "ie", "source_box": "box-0", "target_box": "box-1", "target_enrollment": "enrollment-1"}, |
| 78 | Journal: map[string]string{"pin": string(mustJSON(t, ref)), "prepared": roP, "install_stamp": strconv.FormatUint(c.Holder.Stamp, 10), "install_release": "v1", |
| 79 | "prepared_authority": string(mustJSON(t, auth)), "probing_since": f.now.Format(time.RFC3339Nano), "exec_ack.box-0": `{"op":"rel-1"}`, "exec_ack.box-1": `{"op":"rel-1"}`, |
| 80 | "local_install": string(mustJSON(t, hostclaim.LocalInstall{Box: "box-1", Enrollment: "enrollment-1", InstallStamp: c.Holder.Stamp, JournalSHA256: strings.Repeat("c", 64)}))}} |
| 81 | return nil |
| 82 | }) |
| 83 | if err != nil { |
| 84 | t.Fatal(err) |
| 85 | } |
| 86 | if o, err = hostclaim.Update(ctx, f.claims, o, hostclaim.BeginProbeAttempt(roOp, "box-1")); err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | c8 := &roFiles{} |
| 90 | inputs := &RelocationWakeInputs{ |
| 91 | Launch: func(context.Context, registry.Route) (ClaimServingLaunch, error) { return launch, nil }, |
| 92 | FilesManifest: c8, Markers: plan9D5Markers{bucket}, Claims: f.claims, FilesStore: files, Data: bucket, Bucket: "data", Region: "ie", Box: "box-1", |
| 93 | NewID: f.newID, Now: func() time.Time { return f.now }, Releases: &roReleases{}} |
| 94 | attempts := &relocationProbeAttempts{Claims: f.claims, DataDir: t.TempDir(), Box: "box-1", NewID: f.newID, Now: func() time.Time { return f.now }} |
| 95 | c := f.controller("boot-1", "controller-1", 4101, "start-1") |
| 96 | c.wake.Inputs = ClaimOwnerDispatch{Relocation: &ClaimKindOwners{Inputs: inputs, AdmissionEvidence: inputs.AdmissionEvidence, PublicationEvidence: inputs.PublicationEvidence}} |
| 97 | c.wake.Continue = hostclaim.SessionContinues |
| 98 | // As claimGroupDWiring routes them once the relocation owners bind. |
| 99 | ready := func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil } |
| 100 | c.wake.Operations = claimOperationRouter{hostclaim.KindRelocating: {Attempts: attempts, ProbeReady: ready}} |
| 101 | live, cancel := context.WithCancel(context.Background()) |
| 102 | t.Cleanup(cancel) |
| 103 | done := make(chan struct{}) |
| 104 | go func() { <-live.Done(); close(done) }() |
| 105 | c.wake.mu.Lock() |
| 106 | c.wake.pending[recoveryHost] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done} |
| 107 | c.wake.mu.Unlock() |
| 108 | route := registry.Route{Host: recoveryHost, Kind: "instance", App: "jam", Account: "bab", Status: "relocating", HoldOp: roOp, Socket: filepath.Join(f.guard.DataDir, "app.sock"), DBPath: filepath.Join(f.guard.DataDir, "state", "data.db")} |
| 109 | return &relocationProbeRig{f: f, bucket: bucket, files: files, c8: c8, inputs: inputs, c: c, h: h, o: o, im: im, auth: auth, route: route} |
| 110 | } |
| 111 | |
| 112 | // Finding 6, ruling 47: P's first probe publishes through the shared |
| 113 | // assembly. The claim's own runtime (the recovery fixture's controller) |
| 114 | // admits P's opening through the relocation dispatch, binds region HA's |
| 115 | // attempt journal to the probe start, and Materialize, under the admitted |
| 116 | // spawn selection, runs moves' C8 Establish and writes P's marker: P as |
| 117 | // the marker's op, the install manifest's pin, prepared_authority's |
| 118 | // digest. Publish (the pre-replication call) finds that marker and |
| 119 | // records the receipt; PublicationEvidence then passes. Publish without |
| 120 | // Materialize's fence refuses. |
| 121 | func TestRelocationFirstProbePublishesThroughTheAssembly(t *testing.T) { |
| 122 | r := newRelocationProbeRig(t, roProbeLaunch()) |
| 123 | ctx := t.Context() |
| 124 | f, bucket, files, c8, inputs, c, h, o, im, auth, route := r.f, r.bucket, r.files, r.c8, r.inputs, r.c, r.h, r.o, r.im, r.auth, r.route |
| 125 | admitted, err := c.runtime.Admit(ctx, route, o, roProbe) |
| 126 | if err != nil { |
| 127 | t.Fatal(err) |
| 128 | } |
| 129 | ah := admitted.Claim.Holder |
| 130 | if ah.Generation != roP || ah.Stamp != o.Claim.Holder.Stamp || ah.WriterBox != "box-1" || !hostclaim.PreparedOpened(admitted.Claim) { |
| 131 | t.Fatalf("P's opening: %+v", ah) |
| 132 | } |
| 133 | s, err := c.runtime.Prepare(ctx, route, admitted, roProbe) |
| 134 | if err != nil { |
| 135 | t.Fatal(err) |
| 136 | } |
| 137 | if s.OpenProbe == nil { |
| 138 | t.Fatal("the probe start carries no attempt journal") |
| 139 | } |
| 140 | if _, err := s.OpenProbe(ctx, f.now); err != nil { |
| 141 | t.Fatal(err) |
| 142 | } |
| 143 | if err := s.Reserve(ctx, 0); err != nil { |
| 144 | t.Fatal(err) |
| 145 | } |
| 146 | if a := s.CurrentAttempt(); a.ID == "" || a.Session != ah.Session || a.ControllerID != "controller-1" { |
| 147 | t.Fatalf("the reservation: %+v", a) |
| 148 | } |
| 149 | if err := s.Materialize(ctx); err != nil { |
| 150 | t.Fatal(err) |
| 151 | } |
| 152 | if len(c8.established) != 1 { |
| 153 | t.Fatalf("C8 established %v", c8.established) |
| 154 | } |
| 155 | markerKey := strings.TrimSuffix(h.Location, "/") + "/generation.json" |
| 156 | body, ok := bucket.Current("data", markerKey) |
| 157 | var m genselect.Marker |
| 158 | if !ok || json.Unmarshal(body, &m) != nil || m.Op != roP || m.Box != "box-1" || m.Host != recoveryHost || m.ManifestSHA256 != auth.SHA256 || m.Pin != manifestPin(im) { |
| 159 | t.Fatalf("P's marker: %s", body) |
| 160 | } |
| 161 | written := bucket.LatestID("data", markerKey) |
| 162 | // The pre-replication publish (ClaimWakeAssembly s.Replicate's first |
| 163 | // call for a prepared admission) finds the marker and records it. |
| 164 | in, err := inputs.Select(ctx, route, o, roProbe) |
| 165 | if err != nil { |
| 166 | t.Fatal(err) |
| 167 | } |
| 168 | if err := inputs.Publish(ctx, admitted, in); err != nil { |
| 169 | t.Fatal(err) |
| 170 | } |
| 171 | if bucket.LatestID("data", markerKey) != written { |
| 172 | t.Fatal("Publish rewrote P's marker") |
| 173 | } |
| 174 | if err := inputs.PublicationEvidence(ctx, admitted); err != nil { |
| 175 | t.Fatalf("publication evidence: %v", err) |
| 176 | } |
| 177 | // A fresh wake inputs (another process) has no fence for the session. |
| 178 | other := &RelocationWakeInputs{Launch: inputs.Launch, FilesManifest: c8, Markers: inputs.Markers, Claims: f.claims, FilesStore: files, Data: bucket, Bucket: "data", Region: "ie", Box: "box-1", |
| 179 | NewID: f.newID, Now: inputs.Now, Releases: inputs.Releases} |
| 180 | if err := other.Publish(ctx, admitted, in); !errors.Is(err, hostbox.ErrChanged) { |
| 181 | t.Fatalf("Publish without Materialize's fence: %v", err) |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | // L18 joint round, final 4, item 1 (ruling R-EXEC-CLAIMVERSION): the |
| 186 | // relocation probe's start names the file its launch source hashed. With |
| 187 | // the production BeforeExec hook set, the activator refuses any claimed |
| 188 | // start whose Staged.Version is not its Version (Service.claimedBuild, |
| 189 | // run on the start the runtime's Prepare returns), so a probe selection |
| 190 | // that drops Staged never starts. The start Prepare hands the activator |
| 191 | // for P's prepared opening carries the launch's Staged, and so does every |
| 192 | // selection: a later cold attempt's and an empty P's. |
| 193 | func TestTheRelocationProbeStartNamesTheBuildItHashed(t *testing.T) { |
| 194 | ctx := t.Context() |
| 195 | launch := roProbeLaunch() |
| 196 | r := newRelocationProbeRig(t, launch) |
| 197 | admitted, err := r.c.runtime.Admit(ctx, r.route, r.o, roProbe) |
| 198 | if err != nil { |
| 199 | t.Fatal(err) |
| 200 | } |
| 201 | s, err := r.c.runtime.Prepare(ctx, r.route, admitted, roProbe) |
| 202 | if err != nil { |
| 203 | t.Fatal(err) |
| 204 | } |
| 205 | // claimedBuild's refusal, as the activator applies it with BeforeExec set. |
| 206 | if s.Staged != launch.Staged || s.Version != launch.Version || s.Staged.Version != s.Version { |
| 207 | t.Fatalf("the probe start names build %q but staged %+v; the launch hashed %+v", s.Version, s.Staged, launch.Staged) |
| 208 | } |
| 209 | |
| 210 | for _, o := range []roOptions{{}, {empty: true}} { |
| 211 | f := newROFixture(t, o) |
| 212 | source := f.inputs.Launch |
| 213 | staged := release.StagedIdentity{Version: "v1", StagedPath: "/opt/carlos/versions/v1/acct/app", Digest: strings.Repeat("b", 64)} |
| 214 | f.inputs.Launch = func(ctx context.Context, rt registry.Route) (ClaimServingLaunch, error) { |
| 215 | l, err := source(ctx, rt) |
| 216 | l.Staged = staged |
| 217 | return l, err |
| 218 | } |
| 219 | first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe) |
| 220 | if err != nil { |
| 221 | t.Fatal(err) |
| 222 | } |
| 223 | if first.Staged != staged || first.Staged.Version != first.Version { |
| 224 | t.Fatalf("empty=%v: the first attempt names build %q but staged %+v", o.empty, first.Version, first.Staged) |
| 225 | } |
| 226 | if o.empty { |
| 227 | continue |
| 228 | } |
| 229 | again, err := f.inputs.Select(ctx, f.route, f.admit(f.read(), first), roProbe) |
| 230 | if err != nil { |
| 231 | t.Fatal(err) |
| 232 | } |
| 233 | if again.Prepared || again.Staged != staged || again.Staged.Version != again.Version { |
| 234 | t.Fatalf("a later attempt (prepared %v) names build %q but staged %+v", again.Prepared, again.Version, again.Staged) |
| 235 | } |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | // Ruling 47: a later attempt's fresh generation, and an empty P's opening, |
| 240 | // publish through main's fenced writer only after their first root. |
| 241 | func TestRelocationLaterAttemptsPublishFenced(t *testing.T) { |
| 242 | ctx := t.Context() |
| 243 | f := newROFixture(t, roOptions{}) |
| 244 | first, err := f.inputs.Select(ctx, f.route, f.read(), roProbe) |
| 245 | if err != nil { |
| 246 | t.Fatal(err) |
| 247 | } |
| 248 | opened := f.admit(f.read(), first) |
| 249 | again, err := f.inputs.Select(ctx, f.route, opened, roProbe) |
| 250 | if err != nil { |
| 251 | t.Fatal(err) |
| 252 | } |
| 253 | later := f.admit(opened, again) |
| 254 | gen := later.Claim.Holder.Generation |
| 255 | synced := 0 |
| 256 | pub := ClaimPublication{Sync: func(context.Context) error { synced++; return nil }, Check: func(context.Context) error { return nil }} |
| 257 | if err := f.inputs.Publish(ctx, later, again); !errors.Is(err, hostbox.ErrChanged) { |
| 258 | t.Fatalf("an unfenced marker for a fresh generation: %v", err) |
| 259 | } |
| 260 | if _, err := f.inputs.PublishFenced(ctx, later, again, pub); err == nil || synced != 1 { |
| 261 | t.Fatalf("a marker before the root: %v synced=%d", err, synced) |
| 262 | } |
| 263 | root := moves.SnapshotKey(moves.GenerationPrefix(later.Claim.Holder.Location, gen), 44) |
| 264 | f.bucket.Put("data", root, ltxHeader(t, 1, 44)) |
| 265 | receipt, err := f.inputs.PublishFenced(ctx, later, again, pub) |
| 266 | if err != nil || receipt.Root.Key != root || receipt.MarkerVersion == "" { |
| 267 | t.Fatalf("publish: %+v %v", receipt, err) |
| 268 | } |
| 269 | if err := f.inputs.PublicationEvidence(ctx, later); err != nil { |
| 270 | t.Fatalf("fresh evidence: %v", err) |
| 271 | } |
| 272 | prepared := again |
| 273 | prepared.Prepared = true |
| 274 | if _, err := f.inputs.PublishFenced(ctx, later, prepared, pub); !errors.Is(err, hostbox.ErrChanged) { |
| 275 | t.Fatalf("a prepared admission through the fenced writer: %v", err) |
| 276 | } |
| 277 | |
| 278 | // An empty P: no snapshot, no root until the first transaction. |
| 279 | e := newROFixture(t, roOptions{empty: true}) |
| 280 | ein, err := e.inputs.Select(ctx, e.route, e.read(), roProbe) |
| 281 | if err != nil { |
| 282 | t.Fatal(err) |
| 283 | } |
| 284 | eopened := e.admit(e.read(), ein) |
| 285 | if _, err := e.inputs.PublishFenced(ctx, eopened, ein, pub); err == nil { |
| 286 | t.Fatal("an empty P published a marker with no root") |
| 287 | } |
| 288 | } |
| 289 | |
amadan