| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "bytes" |
| 7 | "context" |
| 8 | "crypto/ed25519" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "fmt" |
| 12 | "os" |
| 13 | "slices" |
| 14 | "strings" |
| 15 | "sync" |
| 16 | "testing" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/carlosframework/platform/internal/hostactors" |
| 20 | "github.com/carlosframework/platform/internal/hostbox" |
| 21 | "github.com/carlosframework/platform/internal/hostclaim" |
| 22 | execlease "github.com/carlosframework/platform/internal/hostexec" |
| 23 | "github.com/carlosframework/platform/internal/hostfloor" |
| 24 | "github.com/carlosframework/platform/internal/hostops" |
| 25 | "github.com/carlosframework/platform/internal/park" |
| 26 | "github.com/carlosframework/platform/internal/provision" |
| 27 | "github.com/carlosframework/platform/internal/release" |
| 28 | ) |
| 29 | |
| 30 | // ackRefreshLease is a participant controller whose lease renews like the |
| 31 | // root broker's: every renewal rotates the controller token and the lease |
| 32 | // ETag and moves the expiry to renewal time plus the TTL. |
| 33 | type ackRefreshLease struct { |
| 34 | mu sync.Mutex |
| 35 | state execlease.State |
| 36 | n int |
| 37 | } |
| 38 | |
| 39 | func (l *ackRefreshLease) renew(at time.Time) { |
| 40 | l.mu.Lock() |
| 41 | defer l.mu.Unlock() |
| 42 | l.n++ |
| 43 | l.state.ControllerID = fmt.Sprintf("token-%d", l.n) |
| 44 | l.state.LeaseETag = fmt.Sprintf(`"lease-%d"`, l.n) |
| 45 | l.state.Expires = at.Add(execlease.TTL) |
| 46 | } |
| 47 | |
| 48 | func (l *ackRefreshLease) Ready(context.Context) (execlease.State, error) { |
| 49 | l.mu.Lock() |
| 50 | defer l.mu.Unlock() |
| 51 | return l.state, nil |
| 52 | } |
| 53 | |
| 54 | func (l *ackRefreshLease) Close(context.Context) error { return nil } |
| 55 | |
| 56 | const ackRefreshHost = "ackrefresh.bab.example" |
| 57 | |
| 58 | func ackRefreshCheck(t *testing.T, step string, err error) { |
| 59 | t.Helper() |
| 60 | if err != nil { |
| 61 | t.Fatalf("%s: %v", step, err) |
| 62 | } |
| 63 | } |
| 64 | |
| 65 | type ackRefreshEvidence struct { |
| 66 | box, console hostactors.Report |
| 67 | identity []byte |
| 68 | } |
| 69 | |
| 70 | func (e ackRefreshEvidence) BoxReport(context.Context, string, hostactors.Box) (hostactors.Report, error) { |
| 71 | return e.box, nil |
| 72 | } |
| 73 | func (e ackRefreshEvidence) ConsoleReport(context.Context, string, hostactors.Console) (hostactors.Report, error) { |
| 74 | return e.console, nil |
| 75 | } |
| 76 | func (e ackRefreshEvidence) LocalConsole(context.Context) (hostactors.Report, error) { |
| 77 | return e.console, nil |
| 78 | } |
| 79 | func (e ackRefreshEvidence) Identity(context.Context, string) ([]byte, error) { |
| 80 | return e.identity, nil |
| 81 | } |
| 82 | func (e ackRefreshEvidence) Fence(context.Context, hostactors.Observed, string, time.Time) (bool, error) { |
| 83 | return false, nil |
| 84 | } |
| 85 | |
| 86 | // ackRefreshController is the console's production wiring (the rootrunner |
| 87 | // lifecycle test's lifecycleController) on a test clock: control writer, |
| 88 | // mutation check and boundary, ProductionAuthority and Controller. |
| 89 | func ackRefreshController(t *testing.T, raw release.Store, key ed25519.PrivateKey, actor hostbox.Actor, pointer release.SystemPointer, now func() time.Time) (hostops.Controller, *hostops.ProductionAuthority, *hostops.ControlWriter) { |
| 90 | t.Helper() |
| 91 | ctx := t.Context() |
| 92 | signer := release.RootSigner(key) |
| 93 | directory := hostactors.Directory{Store: raw, PublicKey: key.Public().(ed25519.PublicKey), Bucket: actor.ControlBucket} |
| 94 | floor, err := hostfloor.Read(ctx, raw, directory.PublicKey, actor.ControlBucket) |
| 95 | ackRefreshCheck(t, "controller floor", err) |
| 96 | platform := release.AdoptPlatform() |
| 97 | pointer.Current.ConsoleArtifacts = map[string]release.SystemArtifact{platform: {Key: "bootstrap/console/v2.0.0/carlos-" + platform, SHA256: pointer.Current.Artifacts[platform].SHA256}} |
| 98 | signed, err := release.SignRootEnvelope(signer, pointer) |
| 99 | ackRefreshCheck(t, "sign actor release", err) |
| 100 | report := hostactors.Report{Component: "router", GeneratedAt: now(), Actor: actor.Box, Enrollment: actor.Enrollment, InstanceID: actor.Cloud.InstanceID, Release: "v2.0.0", ReleaseTag: "v2.0.0", Platform: platform, BinarySHA256: pointer.Current.Artifacts[platform].SHA256, MovesProtocol: 1, BootID: "external-box-boot", FloorSHA256: floor.Digest, FloorGuardReady: true, HostClaimEnabled: true} |
| 101 | console := report |
| 102 | console.Component, console.Actor, console.InstanceID, console.Enrollment, console.ProcessID = "console", "console", "console", strings.Repeat("d", 32), "console-process" |
| 103 | control, err := hostops.NewControlWriter(raw, directory.PublicKey, now) |
| 104 | ackRefreshCheck(t, "control writer", err) |
| 105 | check, err := hostops.NewProductionMutationCheck(raw, control) |
| 106 | ackRefreshCheck(t, "mutation check", err) |
| 107 | records, err := hostops.NewConditionalRecordStore(raw, directory.PublicKey) |
| 108 | ackRefreshCheck(t, "record boundary", err) |
| 109 | data, err := hostops.NewObservedMutationBoundary(true, records, check, nil) |
| 110 | ackRefreshCheck(t, "mutation boundary", err) |
| 111 | directory.Store = data |
| 112 | authority := &hostops.ProductionAuthority{HomeRegion: "ie", Store: data, Control: control, Directory: directory, ConsoleID: "console", Enrollment: console.Enrollment, Namespace: actor.Namespace, Signer: signer, Now: now, Evidence: ackRefreshEvidence{report, console, signed}} |
| 113 | n := 0 |
| 114 | controller := hostops.Controller{Store: data, Control: control, Authority: authority, Now: now, NewID: func() string { n++; return fmt.Sprintf("ackrefresh-%024d", n) }} |
| 115 | return controller, authority, control |
| 116 | } |
| 117 | |
| 118 | type ackRefreshWorld struct { |
| 119 | store release.Store |
| 120 | directory hostactors.Directory |
| 121 | controller hostops.Controller |
| 122 | control *hostops.ControlWriter |
| 123 | recovery hostops.Recovery |
| 124 | actor hostbox.Actor |
| 125 | rec provision.Record |
| 126 | id string |
| 127 | t0 time.Time |
| 128 | now func() time.Time |
| 129 | set func(time.Time) |
| 130 | } |
| 131 | |
| 132 | // ackRefreshCreate is G10 on a v0.28.4-equivalent console path: the |
| 133 | // production create (registration, StartCreate, RecordEffects) through the |
| 134 | // real hostops controller, authority and control writer on a test clock, |
| 135 | // then the console's one in-request Complete, which runs before any ack and |
| 136 | // gets ErrPending. The request then ends, so nothing renews the create's |
| 137 | // registration. |
| 138 | func ackRefreshCreate(t *testing.T) *ackRefreshWorld { |
| 139 | t.Helper() |
| 140 | ctx := t.Context() |
| 141 | var mu sync.Mutex |
| 142 | at := time.Now().UTC().Truncate(time.Second) |
| 143 | w := &ackRefreshWorld{t0: at} |
| 144 | w.now = func() time.Time { mu.Lock(); defer mu.Unlock(); return at } |
| 145 | w.set = func(next time.Time) { mu.Lock(); at = next; mu.Unlock() } |
| 146 | |
| 147 | store := release.NewDirStore(t.TempDir()) |
| 148 | w.store = store |
| 149 | key := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{1}, ed25519.SeedSize)) |
| 150 | actor := hostbox.Actor{Box: "box", Enrollment: strings.Repeat("a", 32), Namespace: "platform", Region: "ie", ControlBucket: "deployment", Cloud: hostbox.CloudIdentity{Provider: "aws", Account: "111122223333", Region: "eu-west-1", InstanceID: "i-0abc"}} |
| 151 | w.actor = actor |
| 152 | roster := hostactors.Roster{V: 1, HomeRegion: "ie", DeploymentBucket: actor.ControlBucket, Revision: 1, Boxes: map[string]hostactors.Box{actor.Box: {BoxID: actor.Box, Enrollment: actor.Enrollment, Namespace: actor.Namespace, Label: actor.Box, Region: actor.Region, Pool: "default", Cloud: hostactors.Cloud{Provider: actor.Cloud.Provider, Account: actor.Cloud.Account, Region: actor.Cloud.Region, InstanceID: actor.Cloud.InstanceID}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"}}, Consoles: map[string]hostactors.Console{}, Holds: map[string]hostactors.Reservation{}, Activations: map[string]hostactors.Reservation{}} |
| 153 | raw, err := release.SignRootEnvelope(release.RootSigner(key), roster) |
| 154 | ackRefreshCheck(t, "sign roster", err) |
| 155 | _, err = store.Put(ctx, hostactors.Key, raw, "") |
| 156 | ackRefreshCheck(t, "write roster", err) |
| 157 | platform := release.AdoptPlatform() |
| 158 | pointer := release.SystemPointer{Schema: 1, Component: "router", Channel: "stable", Spread: "0s", PromotedAt: w.t0, Current: release.SystemRelease{Version: "v2.0.0", ReleaseTag: "v2.0.0", MovesProtocol: 1, Artifacts: map[string]release.SystemArtifact{platform: {Key: release.SystemArtifactKey("v2.0.0", platform), SHA256: strings.Repeat("a", 64)}}}} |
| 159 | |
| 160 | signer := release.RootSigner(key) |
| 161 | directory := hostactors.Directory{Store: store, PublicKey: key.Public().(ed25519.PublicKey), Bucket: actor.ControlBucket} |
| 162 | w.directory = directory |
| 163 | ackRefreshCheck(t, "publish test deployment floor", hostfloor.Publish(ctx, store, signer, actor.ControlBucket, hostfloor.Floor{Min: 1, MinRelease: "v2.0.0"})) |
| 164 | floor, err := hostfloor.Read(ctx, store, directory.PublicKey, actor.ControlBucket) |
| 165 | ackRefreshCheck(t, "read deployment floor", err) |
| 166 | observedRoster, err := directory.Read(ctx) |
| 167 | ackRefreshCheck(t, "read actor roster", err) |
| 168 | observedRoster.Roster.Publication, observedRoster.Roster.FloorSHA256 = "published", floor.Digest |
| 169 | observedRoster.Roster.Regions = map[string]hostactors.Region{actor.Region: {Account: actor.Cloud.Account, AWSRegion: actor.Cloud.Region, Namespaces: []string{actor.Namespace}, Pools: []string{"default"}}} |
| 170 | observedRoster.Roster.Consoles["console"] = hostactors.Console{Enrollment: strings.Repeat("d", 32), InstanceID: "console", Namespaces: []string{actor.Namespace}, DeploymentBucket: actor.ControlBucket} |
| 171 | _, err = directory.Replace(ctx, signer, observedRoster, observedRoster.Roster) |
| 172 | ackRefreshCheck(t, "publish actors", err) |
| 173 | |
| 174 | controller, authority, control := ackRefreshController(t, store, key, actor, pointer, w.now) |
| 175 | w.controller, w.control = controller, control |
| 176 | rec := provision.Record{Schema: 1, Host: ackRefreshHost, Account: "bab", App: "lifecycle", Channel: "stable", CreatedAt: w.t0} |
| 177 | data := controller.Store |
| 178 | appID, err := control.EnrollNewApp(ctx, rec.Account, rec.App) |
| 179 | ackRefreshCheck(t, "enroll app", err) |
| 180 | before, err := hostclaim.Read(ctx, data, rec.Host) |
| 181 | ackRefreshCheck(t, "read before registration", err) |
| 182 | w.id = "create-" + rec.App |
| 183 | request, endRequest := context.WithCancel(ctx) |
| 184 | start, err := control.RegisterStart(request, rec.Account, rec.App, appID, w.id, hostops.Registration{Kind: "create", Host: rec.Host, Owner: hostops.RegistrationOwner{Console: "console", Enrollment: strings.Repeat("d", 32), Controller: "console-controller"}, AppID: appID, Epoch: 1, Claim: hostops.SavedObservation(before)}, w.now()) |
| 185 | ackRefreshCheck(t, "register create", err) |
| 186 | rec.Acquisition = w.id |
| 187 | requested := rec |
| 188 | requested.CreatedAt = time.Time{} |
| 189 | record, err := hostops.PrepareInstance(requested, rec.CreatedAt, park.PlacementRegion(rec.Region)) |
| 190 | ackRefreshCheck(t, "prepare create record", err) |
| 191 | incarnation, err := record.Incarnation() |
| 192 | ackRefreshCheck(t, "create incarnation", err) |
| 193 | ackRefreshCheck(t, "bind registration", start.Bind(request, rec.CreatedAt, incarnation, w.now())) |
| 194 | createCtx := hostops.WithStartHandle(start.Lifecycle(request), start) |
| 195 | observed, err := authority.StartCreate(createCtx, before, record, w.id, "runner") |
| 196 | ackRefreshCheck(t, "production begin create", err) |
| 197 | ackRefreshCheck(t, "production record effects", authority.RecordEffects(createCtx, observed, record)) |
| 198 | if _, err = controller.Complete(createCtx, rec.Host, w.id); !errors.Is(err, hostops.ErrPending) { |
| 199 | t.Fatalf("in-request Complete before any ack = %v, want ErrPending", err) |
| 200 | } |
| 201 | endRequest() |
| 202 | w.rec = rec |
| 203 | |
| 204 | noop := func(context.Context) error { return nil } |
| 205 | noopObserved := func(context.Context, hostclaim.Observed) error { return nil } |
| 206 | // SweepAdmissions settles registrations as the console's sweep does once |
| 207 | // no exclusive operation is open (console claim_runtime.go). |
| 208 | sweep := func(ctx context.Context) error { |
| 209 | apps, err := hostops.Apps(ctx, store) |
| 210 | if err != nil { |
| 211 | return err |
| 212 | } |
| 213 | for _, app := range apps { |
| 214 | a, err := hostops.ReadCanonicalAdmission(ctx, store, app.Account, app.Name) |
| 215 | if errors.Is(err, os.ErrNotExist) { |
| 216 | continue |
| 217 | } |
| 218 | if err != nil { |
| 219 | return err |
| 220 | } |
| 221 | if a.Value.Op != nil { |
| 222 | return fmt.Errorf("unexpected exclusive op %s", a.Value.Op.ID) |
| 223 | } |
| 224 | for id := range a.Value.Registrations { |
| 225 | if err := control.SettleRegistration(ctx, app.Account, app.Name, a.Value.AppID, id, w.now()); err != nil && !errors.Is(err, hostops.ErrPending) { |
| 226 | return err |
| 227 | } |
| 228 | } |
| 229 | } |
| 230 | return nil |
| 231 | } |
| 232 | w.recovery = hostops.Recovery{Controller: controller, Purge: noopObserved, Placement: noopObserved, SweepRecords: noop, SweepPurges: noop, SweepAdmissions: sweep} |
| 233 | return w |
| 234 | } |
| 235 | |
| 236 | func (w *ackRefreshWorld) claim(t *testing.T) hostclaim.Observed { |
| 237 | t.Helper() |
| 238 | o, err := hostclaim.Read(t.Context(), w.store, w.rec.Host) |
| 239 | if err != nil || o.Absent || o.Claim == nil { |
| 240 | t.Fatalf("read claim: %+v %v", o, err) |
| 241 | } |
| 242 | return o |
| 243 | } |
| 244 | |
| 245 | func (w *ackRefreshWorld) ack(t *testing.T) hostclaim.ExecutionAcknowledgement { |
| 246 | t.Helper() |
| 247 | var ack hostclaim.ExecutionAcknowledgement |
| 248 | if err := json.Unmarshal([]byte(w.claim(t).Claim.Op.Journal["exec_ack."+w.actor.Box]), &ack); err != nil { |
| 249 | t.Fatal("missing acknowledgement", err) |
| 250 | } |
| 251 | return ack |
| 252 | } |
| 253 | |
| 254 | // registration is the create's admission registration, which the stuck |
| 255 | // harness still listed, expired. |
| 256 | func (w *ackRefreshWorld) registration(t *testing.T) (hostops.Registration, bool) { |
| 257 | t.Helper() |
| 258 | a, err := hostops.ReadCanonicalAdmission(t.Context(), w.store, w.rec.Account, w.rec.App) |
| 259 | if err != nil { |
| 260 | t.Fatal(err) |
| 261 | } |
| 262 | reg, ok := a.Value.Registrations[w.id] |
| 263 | return reg, ok |
| 264 | } |
| 265 | |
| 266 | // participant is the box's participation for the create, holding lease as |
| 267 | // its controller in a slot it opened. |
| 268 | func (w *ackRefreshWorld) participant(t *testing.T, lease *ackRefreshLease) *productionClaimLocal { |
| 269 | t.Helper() |
| 270 | o := w.claim(t) |
| 271 | if o.Claim.Op.Kind != hostclaim.KindCreating || o.Claim.Op.ID != w.id || !slices.Contains(o.Claim.Op.Participants, w.actor.Box) { |
| 272 | t.Fatalf("create is not waiting on the box: %+v", o.Claim.Op) |
| 273 | } |
| 274 | lease.state = execlease.State{V: 1, State: "active", Binding: execlease.Binding{Box: w.actor.Box, Enrollment: w.actor.Enrollment, Host: w.rec.Host, Incarnation: o.Claim.Holder.Incarnation, Op: w.id, BootID: "boot-1"}} |
| 275 | ready := make(chan struct{}) |
| 276 | close(ready) |
| 277 | return &productionClaimLocal{Resources: ClaimRuntimeResources{Mode: "agent"}, Store: w.store, now: w.now, slots: map[string]*claimParticipation{w.rec.Host: {Ready: ready, Controller: lease}}} |
| 278 | } |
| 279 | |
| 280 | func (w *ackRefreshWorld) pass(t *testing.T, p *productionClaimLocal) { |
| 281 | t.Helper() |
| 282 | if err := p.acknowledgeOperation(t.Context(), w.actor, w.claim(t)); err != nil { |
| 283 | t.Fatal("participation pass", err) |
| 284 | } |
| 285 | } |
| 286 | |
| 287 | // TestClaimAckRefreshCompletesCreateThroughRecovery is G10 end to end on |
| 288 | // the console's recovery path: the box's ack lands 15 s after the op starts, |
| 289 | // the box keeps it fresh on its 30 s participation tick, and the console's |
| 290 | // recovery pass (RecoveryPass -> RecoverCreate -> Complete -> Executor -> |
| 291 | // FirstPhase) completes the create at started_at + 5 min. By then the create |
| 292 | // registration has expired but is still listed, both effects are complete |
| 293 | // and the roster hold is still present. Without the refresh, recovery gets |
| 294 | // ErrPending and the create stays stuck, as on the harness. |
| 295 | func TestClaimAckRefreshCompletesCreateThroughRecovery(t *testing.T) { |
| 296 | for _, refresh := range []bool{true, false} { |
| 297 | name := "refreshed" |
| 298 | if !refresh { |
| 299 | name = "first ack only" |
| 300 | } |
| 301 | t.Run(name, func(t *testing.T) { |
| 302 | w := ackRefreshCreate(t) |
| 303 | lease := &ackRefreshLease{} |
| 304 | box := w.participant(t, lease) |
| 305 | first := w.t0.Add(15 * time.Second) |
| 306 | w.set(first) |
| 307 | lease.renew(first) |
| 308 | w.pass(t, box) |
| 309 | ack0 := w.ack(t) |
| 310 | if ack0.ControllerID != "token-1" || !ack0.Expires.Equal(first.Add(execlease.TTL)) { |
| 311 | t.Fatalf("first ack %+v", ack0) |
| 312 | } |
| 313 | deadline := w.t0.Add(5*time.Minute + time.Second) |
| 314 | for tick := first.Add(30 * time.Second); ; tick = tick.Add(30 * time.Second) { |
| 315 | if tick.After(deadline) { |
| 316 | tick = deadline |
| 317 | } |
| 318 | w.set(tick) |
| 319 | // The worst case for the box: its last renewal was a full |
| 320 | // renewal interval ago. |
| 321 | lease.renew(tick.Add(-execlease.Renewal)) |
| 322 | if refresh { |
| 323 | // Usability only shrinks with time, so an ack still usable |
| 324 | // at each tick, before that tick's pass, never lapsed. |
| 325 | if ack := w.ack(t); !w.now().Before(ack.Expires.Add(-hostops.RegistrationSkew)) { |
| 326 | t.Fatalf("ack lapsed before the pass at +%s: %+v", tick.Sub(w.t0), ack) |
| 327 | } |
| 328 | w.pass(t, box) |
| 329 | ack := w.ack(t) |
| 330 | if !w.now().Before(ack.Expires.Add(-hostops.RegistrationSkew)) { |
| 331 | t.Fatalf("ack unusable at +%s: %+v", tick.Sub(w.t0), ack) |
| 332 | } |
| 333 | if ack.Op != ack0.Op || ack.Enrollment != ack0.Enrollment || ack.BootID != ack0.BootID || ack.ControllerID != ack0.ControllerID { |
| 334 | t.Fatalf("refresh rebound the ack: %+v, first %+v", ack, ack0) |
| 335 | } |
| 336 | } |
| 337 | if tick.Equal(deadline) { |
| 338 | break |
| 339 | } |
| 340 | // Recovery runs before the op is five minutes old: it neither |
| 341 | // completes nor cancels it, and the sweep keeps the registration. |
| 342 | if err := w.control.RecoveryPass(t.Context(), w.recovery); err != nil { |
| 343 | t.Fatalf("recovery pass at +%s: %v", tick.Sub(w.t0), err) |
| 344 | } |
| 345 | if o := w.claim(t); o.Claim.Op.Kind != hostclaim.KindCreating || o.Claim.Op.ID != w.id { |
| 346 | t.Fatalf("op left creating at +%s: %+v", tick.Sub(w.t0), o.Claim.Op) |
| 347 | } |
| 348 | if _, ok := w.registration(t); !ok { |
| 349 | t.Fatalf("sweep removed the create registration at +%s", tick.Sub(w.t0)) |
| 350 | } |
| 351 | } |
| 352 | if refresh && w.ack(t).Expires.Equal(ack0.Expires) { |
| 353 | t.Fatal("ack never refreshed") |
| 354 | } |
| 355 | if !refresh && w.ack(t) != ack0 { |
| 356 | t.Fatal("first-ack-only box changed its ack") |
| 357 | } |
| 358 | |
| 359 | // The harness's state at recovery time. |
| 360 | o := w.claim(t) |
| 361 | reg, ok := w.registration(t) |
| 362 | if !ok || reg.Kind != "create" || !reg.Claim.Absent || hostops.RegistrationLive(reg, w.now()) { |
| 363 | t.Fatalf("want an expired, still-listed create registration: %+v %v", reg, ok) |
| 364 | } |
| 365 | adm, err := hostops.ReadCanonicalAdmission(t.Context(), w.store, w.rec.Account, w.rec.App) |
| 366 | if err != nil { |
| 367 | t.Fatal(err) |
| 368 | } |
| 369 | effects := 0 |
| 370 | for _, e := range adm.Value.Effects { |
| 371 | if e.OperationID == w.id { |
| 372 | effects++ |
| 373 | if e.State != "complete" { |
| 374 | t.Fatalf("effect not complete: %+v", e) |
| 375 | } |
| 376 | } |
| 377 | } |
| 378 | if effects != 2 { |
| 379 | t.Fatalf("create effects = %d, want 2", effects) |
| 380 | } |
| 381 | roster, err := w.directory.Read(t.Context()) |
| 382 | if err != nil { |
| 383 | t.Fatal(err) |
| 384 | } |
| 385 | if hold, ok := roster.Roster.Holds[o.Claim.Op.Intent["actors_hold"]]; !ok || hold.OpID != w.id { |
| 386 | t.Fatalf("roster hold missing: %+v", roster.Roster.Holds) |
| 387 | } |
| 388 | |
| 389 | if err := w.control.RecoveryPass(t.Context(), w.recovery); err != nil { |
| 390 | t.Fatal("recovery pass", err) |
| 391 | } |
| 392 | o = w.claim(t) |
| 393 | if !refresh { |
| 394 | if o.Claim.Op.Kind != hostclaim.KindCreating || o.Claim.Op.ID != w.id { |
| 395 | t.Fatalf("expired ack completed or cancelled the create: %+v", o.Claim.Op) |
| 396 | } |
| 397 | if err := w.controller.RecoverCreate(t.Context(), w.rec.Host); !errors.Is(err, hostops.ErrPending) { |
| 398 | t.Fatalf("RecoverCreate with an expired ack = %v, want ErrPending", err) |
| 399 | } |
| 400 | if _, ok := w.registration(t); !ok { |
| 401 | t.Fatal("stuck create lost its registration") |
| 402 | } |
| 403 | return |
| 404 | } |
| 405 | if o.Claim.Op.Kind != hostclaim.KindServing || o.Claim.Holder == nil || o.Claim.Holder.Stamp == 0 { |
| 406 | t.Fatalf("recovery did not complete the create: %+v %+v", o.Claim.Op, o.Claim.Holder) |
| 407 | } |
| 408 | if _, ok := w.registration(t); ok { |
| 409 | t.Fatal("completed create kept its registration") |
| 410 | } |
| 411 | }) |
| 412 | } |
| 413 | } |
| 414 | |
| 415 | // TestClaimAckRefreshLeavesOtherControllersAlone: the box refreshes only an |
| 416 | // ack its own controller wrote in this boot. A restarted agent (a new slot) |
| 417 | // or another boot leaves the old ack to expire, and the ack is not touched |
| 418 | // while more than ackRefreshBefore remains. |
| 419 | func TestClaimAckRefreshLeavesOtherControllersAlone(t *testing.T) { |
| 420 | for _, tc := range []string{"not yet due", "restarted controller", "other boot", "lease not renewed", "foreign enrollment", "foreign op", "foreign controller", "lease for another incarnation"} { |
| 421 | t.Run(tc, func(t *testing.T) { |
| 422 | w := ackRefreshCreate(t) |
| 423 | lease := &ackRefreshLease{} |
| 424 | box := w.participant(t, lease) |
| 425 | first := w.t0.Add(15 * time.Second) |
| 426 | w.set(first) |
| 427 | lease.renew(first) |
| 428 | w.pass(t, box) |
| 429 | ack0 := w.ack(t) |
| 430 | at := ack0.Expires.Add(-ackRefreshBefore + time.Second) |
| 431 | switch tc { |
| 432 | case "not yet due": |
| 433 | at = ack0.Expires.Add(-ackRefreshBefore) |
| 434 | case "restarted controller": |
| 435 | box = w.participant(t, lease) |
| 436 | lease.state.Expires = ack0.Expires |
| 437 | case "other boot": |
| 438 | lease.mu.Lock() |
| 439 | lease.state.BootID = "boot-2" |
| 440 | lease.mu.Unlock() |
| 441 | case "foreign enrollment", "foreign op", "foreign controller": |
| 442 | // A journalled ack this slot did not write, or one naming |
| 443 | // another enrollment or op, is never extended. |
| 444 | foreign := ack0 |
| 445 | switch tc { |
| 446 | case "foreign op": |
| 447 | foreign.Op = "other-op" |
| 448 | case "foreign enrollment": |
| 449 | foreign.Enrollment = strings.Repeat("b", 32) |
| 450 | default: |
| 451 | foreign.ControllerID = "earlier-controller" |
| 452 | } |
| 453 | raw, err := json.Marshal(foreign) |
| 454 | if err != nil { |
| 455 | t.Fatal(err) |
| 456 | } |
| 457 | if _, err = hostclaim.Update(t.Context(), w.store, w.claim(t), func(c *hostclaim.Claim) error { |
| 458 | c.Op.Journal["exec_ack."+w.actor.Box] = string(raw) |
| 459 | return nil |
| 460 | }); err != nil { |
| 461 | t.Fatal(err) |
| 462 | } |
| 463 | ack0 = foreign |
| 464 | } |
| 465 | w.set(at) |
| 466 | if tc != "lease not renewed" { |
| 467 | lease.renew(at) |
| 468 | } |
| 469 | if tc == "lease for another incarnation" { |
| 470 | lease.mu.Lock() |
| 471 | lease.state.Incarnation = "other-incarnation" |
| 472 | lease.mu.Unlock() |
| 473 | if err := box.acknowledgeOperation(t.Context(), w.actor, w.claim(t)); !errors.Is(err, execlease.ErrLost) { |
| 474 | t.Fatalf("refresh from a lease for another incarnation = %v, want ErrLost", err) |
| 475 | } |
| 476 | } else { |
| 477 | w.pass(t, box) |
| 478 | } |
| 479 | if got := w.ack(t); got != ack0 { |
| 480 | t.Fatalf("ack changed: %+v, was %+v", got, ack0) |
| 481 | } |
| 482 | if tc == "not yet due" { |
| 483 | w.set(at.Add(time.Second)) |
| 484 | w.pass(t, box) |
| 485 | if got := w.ack(t); got.Expires.Equal(ack0.Expires) || got.ControllerID != ack0.ControllerID { |
| 486 | t.Fatalf("due ack not refreshed: %+v", got) |
| 487 | } |
| 488 | } |
| 489 | }) |
| 490 | } |
| 491 | } |
| 492 | |
| 493 | // TestClaimAckRefreshProductionClient runs the refresh against the real root |
| 494 | // broker and RootClient. A renewal rotates the controller token, so the |
| 495 | // refreshed ack keeps the first ack's controller ID and carries the renewed |
| 496 | // lease ETag and expiry. |
| 497 | func TestClaimAckRefreshProductionClient(t *testing.T) { |
| 498 | if testing.Short() { |
| 499 | t.Skip("waits for one real lease renewal") |
| 500 | } |
| 501 | f, root, _, _, local := productionExecutionFixture(t, hostclaim.KindLanding) |
| 502 | before := f.read() |
| 503 | h := before.Claim.Holder |
| 504 | rec := provision.Record{Schema: 1, Host: before.Host, Account: h.Account, App: h.App, Channel: "stable", Region: "ie", CreatedAt: f.now} |
| 505 | body, err := json.Marshal(rec) |
| 506 | if err != nil { |
| 507 | t.Fatal(err) |
| 508 | } |
| 509 | _, err = hostclaim.Update(t.Context(), f.claims, before, func(c *hostclaim.Claim) error { |
| 510 | *c = hostclaim.Claim{V: c.V, Host: c.Host, Gen: c.Gen} |
| 511 | return hostclaim.BeginSelectedCreate(hostclaim.NewHolder{Host: rec.Host, Account: rec.Account, App: rec.App, Family: "instance", Key: h.Key, Region: "ie", RecordRegion: "ie", CreatedAt: rec.CreatedAt}, body, "create", "test", f.now, "box-1", "en", "app-id", 1, "own", "record")(c) |
| 512 | }) |
| 513 | if err != nil { |
| 514 | t.Fatal(err) |
| 515 | } |
| 516 | actor, err := root.InventoryActor(t.Context()) |
| 517 | if err != nil { |
| 518 | t.Fatal(err) |
| 519 | } |
| 520 | ack := func() hostclaim.ExecutionAcknowledgement { |
| 521 | var a hostclaim.ExecutionAcknowledgement |
| 522 | if err := json.Unmarshal([]byte(f.read().Claim.Op.Journal["exec_ack."+actor.Box]), &a); err != nil { |
| 523 | t.Fatal("missing acknowledgement", err) |
| 524 | } |
| 525 | return a |
| 526 | } |
| 527 | if err = local.participateHost(t.Context(), actor, rec.Host); err != nil { |
| 528 | t.Fatal("participate and acknowledge", err) |
| 529 | } |
| 530 | ack0 := ack() |
| 531 | controller := local.slots[rec.Host].Controller |
| 532 | var state execlease.State |
| 533 | for deadline := time.Now().Add(2 * execlease.Renewal); ; { |
| 534 | if state, err = controller.Ready(t.Context()); err != nil { |
| 535 | t.Fatal(err) |
| 536 | } |
| 537 | if state.Expires.After(ack0.Expires) { |
| 538 | break |
| 539 | } |
| 540 | if time.Now().After(deadline) { |
| 541 | t.Fatal("lease never renewed") |
| 542 | } |
| 543 | time.Sleep(250 * time.Millisecond) |
| 544 | } |
| 545 | if state.ControllerID == ack0.ControllerID || state.BootID != ack0.BootID { |
| 546 | t.Fatalf("renewal did not rotate the controller token: %+v, ack %+v", state, ack0) |
| 547 | } |
| 548 | // Past the first ack's usable window, as at the console's recovery. |
| 549 | local.now = func() time.Time { return ack0.Expires.Add(-hostops.RegistrationSkew) } |
| 550 | if err = local.participateHost(t.Context(), actor, rec.Host); err != nil { |
| 551 | t.Fatal("refreshing pass", err) |
| 552 | } |
| 553 | got := ack() |
| 554 | if state, err = controller.Ready(t.Context()); err != nil { |
| 555 | t.Fatal(err) |
| 556 | } |
| 557 | if got.Op != ack0.Op || got.Enrollment != ack0.Enrollment || got.BootID != ack0.BootID || got.ControllerID != ack0.ControllerID || got.LeaseETag == "" || got.LeaseETag == ack0.LeaseETag || !got.Expires.After(ack0.Expires) || got.Expires.After(state.Expires) { |
| 558 | t.Fatalf("refreshed ack %+v, first %+v, lease %+v", got, ack0, state) |
| 559 | } |
| 560 | } |
| 561 | |
amadan