| 1 | package hostops |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/ed25519" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "strconv" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "testing" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/release" |
| 15 | ) |
| 16 | |
| 17 | // hookStore runs hook once, on the first admission Put whose next value |
| 18 | // has an effect in state want. A nil hook error lets the Put through; on a |
| 19 | // hook error, land decides whether the Put still reaches the store (a lost |
| 20 | // response) or not. |
| 21 | type hookStore struct { |
| 22 | *release.DirStore |
| 23 | mu sync.Mutex |
| 24 | want string |
| 25 | land bool |
| 26 | hook func(ctx context.Context) error |
| 27 | } |
| 28 | |
| 29 | func (s *hookStore) Put(ctx context.Context, key string, body []byte, tag string) (string, error) { |
| 30 | if hook := s.take(key, body); hook != nil { |
| 31 | herr := hook(ctx) |
| 32 | if herr == nil { |
| 33 | return s.DirStore.Put(ctx, key, body, tag) |
| 34 | } |
| 35 | if !s.land { |
| 36 | return "", herr |
| 37 | } |
| 38 | if _, err := s.DirStore.Put(context.WithoutCancel(ctx), key, body, tag); err != nil { |
| 39 | return "", err |
| 40 | } |
| 41 | return "", herr |
| 42 | } |
| 43 | return s.DirStore.Put(ctx, key, body, tag) |
| 44 | } |
| 45 | |
| 46 | func (s *hookStore) take(key string, body []byte) func(context.Context) error { |
| 47 | s.mu.Lock() |
| 48 | defer s.mu.Unlock() |
| 49 | if s.hook == nil || !strings.HasSuffix(key, "/admission.json") { |
| 50 | return nil |
| 51 | } |
| 52 | var a AppAdmission |
| 53 | if json.Unmarshal(body, &a) != nil { |
| 54 | return nil |
| 55 | } |
| 56 | for _, e := range a.Effects { |
| 57 | if e.State == s.want { |
| 58 | hook := s.hook |
| 59 | s.hook = nil |
| 60 | return hook |
| 61 | } |
| 62 | } |
| 63 | return nil |
| 64 | } |
| 65 | |
| 66 | type mutationFixture struct { |
| 67 | raw *hookStore |
| 68 | w *ControlWriter |
| 69 | data release.Store |
| 70 | now time.Time |
| 71 | owner RegistrationOwner |
| 72 | } |
| 73 | |
| 74 | func newMutationFixture(t *testing.T) *mutationFixture { |
| 75 | t.Helper() |
| 76 | f := &mutationFixture{raw: &hookStore{DirStore: release.NewDirStore(t.TempDir())}, now: time.Date(2026, 10, 7, 12, 0, 0, 0, time.UTC)} |
| 77 | pub, _, err := ed25519.GenerateKey(nil) |
| 78 | if err != nil { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | if f.w, err = NewControlWriter(f.raw, pub, func() time.Time { return f.now }); err != nil { |
| 82 | t.Fatal(err) |
| 83 | } |
| 84 | check, err := NewProductionMutationCheck(f.raw, f.w) |
| 85 | if err != nil { |
| 86 | t.Fatal(err) |
| 87 | } |
| 88 | if f.data, err = NewMutationBoundary(true, f.raw, check); err != nil { |
| 89 | t.Fatal(err) |
| 90 | } |
| 91 | f.owner = RegistrationOwner{Console: "console", Enrollment: "enrollment", Controller: "controller"} |
| 92 | if err = BindMutationRunner(f.data, f.w, f.owner, func(context.Context, string, string) (string, error) { return "app", nil }); err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | seedAdmission(t, f.raw, "bab", "demo", openAdmission("app")) |
| 96 | return f |
| 97 | } |
| 98 | |
| 99 | func (f *mutationFixture) admission(t *testing.T) AppAdmission { |
| 100 | t.Helper() |
| 101 | a, err := ReadAdmission(t.Context(), f.raw, "bab", "demo", "app") |
| 102 | if err != nil { |
| 103 | t.Fatal(err) |
| 104 | } |
| 105 | return a.Value |
| 106 | } |
| 107 | |
| 108 | func onlyEffect(t *testing.T, a AppAdmission) Effect { |
| 109 | t.Helper() |
| 110 | if len(a.Effects) != 1 { |
| 111 | t.Fatalf("want one effect, got %+v", a.Effects) |
| 112 | } |
| 113 | for _, e := range a.Effects { |
| 114 | return e |
| 115 | } |
| 116 | return Effect{} |
| 117 | } |
| 118 | |
| 119 | // Review P1-1: Issue failing after BeginExclusive (cancelled request, |
| 120 | // create freeze, lost issued-CAS response) must not leave the app closed |
| 121 | // when nothing was issued; the caller sees the original error, unchanged in |
| 122 | // kind. An issued effect is never settled by the issuer's recovery. |
| 123 | func TestRunAppMutationIssueFailureRecovers(t *testing.T) { |
| 124 | for _, tc := range []struct { |
| 125 | name string |
| 126 | want string // effect state that triggers the hook |
| 127 | land bool |
| 128 | freeze bool |
| 129 | wantErr error |
| 130 | reopened bool |
| 131 | effect string // "" none, else the effect's final state/result |
| 132 | }{ |
| 133 | {name: "cancelled before prepare", want: "prepared", wantErr: context.Canceled, reopened: true}, |
| 134 | {name: "cancelled after prepare", want: "issued", wantErr: context.Canceled, reopened: true, effect: "complete/superseded"}, |
| 135 | {name: "create freeze", freeze: true, wantErr: ErrIdentity, reopened: true}, |
| 136 | {name: "issued response lost", want: "issued", land: true, wantErr: context.Canceled, effect: "issued/"}, |
| 137 | } { |
| 138 | t.Run(tc.name, func(t *testing.T) { |
| 139 | f := newMutationFixture(t) |
| 140 | ctx, cancel := context.WithCancel(t.Context()) |
| 141 | defer cancel() |
| 142 | if tc.freeze { |
| 143 | // An unverifiable backfill plan: CheckCreateFreeze refuses. |
| 144 | if _, err := f.raw.DirStore.Put(ctx, BackfillKey, []byte(`{}`), ""); err != nil { |
| 145 | t.Fatal(err) |
| 146 | } |
| 147 | } else { |
| 148 | f.raw.want, f.raw.land = tc.want, tc.land |
| 149 | f.raw.hook = func(context.Context) error { cancel(); return context.Canceled } |
| 150 | } |
| 151 | called := false |
| 152 | err := release.WithAppMutation(ctx, f.data, "bab", "demo", "test", func(context.Context) error { called = true; return nil }) |
| 153 | if called { |
| 154 | t.Fatal("callback ran without an issued effect") |
| 155 | } |
| 156 | if !errors.Is(err, tc.wantErr) { |
| 157 | t.Fatalf("original error lost: %v", err) |
| 158 | } |
| 159 | if !errors.Is(tc.wantErr, ErrPending) && errors.Is(err, ErrPending) { |
| 160 | t.Fatalf("recovery outcome reclassified the error as pending: %v", err) |
| 161 | } |
| 162 | a := f.admission(t) |
| 163 | if tc.reopened { |
| 164 | if a.State != "open" || a.Op != nil || a.Epoch != 2 { |
| 165 | t.Fatalf("unissued operation wedged the app: %+v", a) |
| 166 | } |
| 167 | } else if a.State != "closed" || a.Op == nil || a.Epoch != 1 { |
| 168 | t.Fatalf("issued effect lost its operation: %+v", a) |
| 169 | } |
| 170 | switch { |
| 171 | case tc.effect == "" && len(a.Effects) != 0: |
| 172 | t.Fatalf("unexpected effects: %+v", a.Effects) |
| 173 | case tc.effect != "": |
| 174 | if e := onlyEffect(t, a); e.State+"/"+e.ResultReference != tc.effect { |
| 175 | t.Fatalf("effect %+v, want %s", e, tc.effect) |
| 176 | } |
| 177 | } |
| 178 | if !tc.reopened { |
| 179 | // Neither the sweep, at any age, may settle the issued effect. |
| 180 | f.now = f.now.Add(24 * time.Hour) |
| 181 | if err := f.w.RecoverMutation(t.Context(), "bab", "demo", "app", a.Op.ID); !errors.Is(err, ErrPending) { |
| 182 | t.Fatalf("issued effect recovered: %v", err) |
| 183 | } |
| 184 | } |
| 185 | }) |
| 186 | } |
| 187 | } |
| 188 | |
| 189 | // Review P1-1: recovery between a live issuer's prepare and issued CASes |
| 190 | // supersedes the prepared effect and reopens with a new epoch; the late |
| 191 | // issued CAS is refused, the callback never runs, and a fresh mutation |
| 192 | // proceeds. |
| 193 | func TestLateIssueIsFencedByUnissuedRecovery(t *testing.T) { |
| 194 | f := newMutationFixture(t) |
| 195 | ctx := t.Context() |
| 196 | var opID string |
| 197 | f.raw.want = "issued" |
| 198 | f.raw.hook = func(context.Context) error { |
| 199 | a := f.admission(t) |
| 200 | opID = a.Op.ID |
| 201 | // Too young: the sweep leaves a live issuer alone. |
| 202 | if err := f.w.RecoverMutation(ctx, "bab", "demo", "app", opID); !errors.Is(err, ErrPending) { |
| 203 | t.Fatalf("recovered a young operation: %v", err) |
| 204 | } |
| 205 | f.now = f.now.Add(UnissuedMutationRecoverAfter + time.Second) |
| 206 | if err := f.w.RecoverMutation(ctx, "bab", "demo", "app", opID); err != nil { |
| 207 | t.Fatalf("aged unissued operation not recovered: %v", err) |
| 208 | } |
| 209 | return nil // the issuer's issued Put proceeds, against a stale etag |
| 210 | } |
| 211 | called := false |
| 212 | err := release.WithAppMutation(ctx, f.data, "bab", "demo", "test", func(context.Context) error { called = true; return nil }) |
| 213 | if called || !errors.Is(err, ErrPending) { |
| 214 | t.Fatalf("late issuer was not fenced: called=%v err=%v", called, err) |
| 215 | } |
| 216 | a := f.admission(t) |
| 217 | if a.State != "open" || a.Op != nil || a.Epoch != 2 { |
| 218 | t.Fatalf("not reopened once: %+v", a) |
| 219 | } |
| 220 | if e := a.Effects[opID]; e.State != "complete" || e.ResultReference != "superseded" { |
| 221 | t.Fatalf("prepared effect not superseded: %+v", e) |
| 222 | } |
| 223 | if err = release.WithAppMutation(ctx, f.data, "bab", "demo", "test", func(context.Context) error { return nil }); err != nil { |
| 224 | t.Fatalf("app stayed wedged: %v", err) |
| 225 | } |
| 226 | } |
| 227 | |
| 228 | // Review P1-1: a lost-response begin (or a crash right after it) leaves an |
| 229 | // operation with no effect; the sweep reopens it once it is old enough, and |
| 230 | // the stale handle can neither issue nor finish. |
| 231 | func TestRecoverMutationAbandonsEffectlessGenericOperation(t *testing.T) { |
| 232 | f := newMutationFixture(t) |
| 233 | ctx := t.Context() |
| 234 | started := strconv.FormatInt(f.now.Unix(), 10) |
| 235 | h, err := f.w.BeginExclusive(ctx, "bab", "demo", "app", AppOperation{ID: "op", Kind: OpAppMutation, AppID: "app", Intent: map[string]string{"action": "test", MutationActionStarted: started}}, f.owner) |
| 236 | if err != nil { |
| 237 | t.Fatal(err) |
| 238 | } |
| 239 | if err = f.w.RecoverMutation(ctx, "bab", "demo", "app", "op"); !errors.Is(err, ErrPending) { |
| 240 | t.Fatalf("young operation recovered: %v", err) |
| 241 | } |
| 242 | f.now = f.now.Add(UnissuedMutationRecoverAfter) |
| 243 | if err = f.w.RecoverMutation(ctx, "bab", "demo", "app", "op"); err != nil { |
| 244 | t.Fatal(err) |
| 245 | } |
| 246 | if a := f.admission(t); a.State != "open" || a.Op != nil || a.Epoch != 2 || len(a.Effects) != 0 { |
| 247 | t.Fatalf("not reopened: %+v", a) |
| 248 | } |
| 249 | if _, _, err = h.Runner(func() time.Time { return f.now }).Issue(ctx, Effect{ID: "op", Resource: OpAppMutation, Action: "test"}); !errors.Is(err, ErrPending) { |
| 250 | t.Fatalf("stale handle issued: %v", err) |
| 251 | } |
| 252 | if err = h.FinishMutation(ctx); !errors.Is(err, ErrPending) { |
| 253 | t.Fatalf("stale handle finished: %v", err) |
| 254 | } |
| 255 | if a := f.admission(t); a.State != "open" || a.Epoch != 2 || len(a.Effects) != 0 { |
| 256 | t.Fatalf("stale handle changed the admission: %+v", a) |
| 257 | } |
| 258 | } |
| 259 | |
| 260 | // Generic recovery never settles issued or uncertain work by age, never |
| 261 | // touches a registration-holding app, and leaves restore and mail to their |
| 262 | // specialized branches. |
| 263 | func TestRecoverMutationGenericRefusals(t *testing.T) { |
| 264 | old := strconv.FormatInt(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), 10) |
| 265 | for _, tc := range []struct { |
| 266 | name string |
| 267 | action string |
| 268 | mutate func(*AppAdmission) |
| 269 | }{ |
| 270 | {"issued", "test", func(a *AppAdmission) { |
| 271 | a.Effects["op"] = Effect{ID: "op", OperationID: "op", AppID: "app", Resource: OpAppMutation, Action: "test", State: "issued"} |
| 272 | }}, |
| 273 | {"uncertain", "test", func(a *AppAdmission) { |
| 274 | a.Effects["op"] = Effect{ID: "op", OperationID: "op", AppID: "app", Resource: OpAppMutation, Action: "test", State: "uncertain"} |
| 275 | }}, |
| 276 | {"sibling uncertain", "test", func(a *AppAdmission) { |
| 277 | a.Effects["op"] = Effect{ID: "op", OperationID: "op", AppID: "app", Resource: OpAppMutation, Action: "test", State: "prepared"} |
| 278 | a.Effects["x"] = Effect{ID: "x", OperationID: "op", AppID: "app", Resource: "other", Action: "test", State: "uncertain"} |
| 279 | }}, |
| 280 | {"foreign prepared", "test", func(a *AppAdmission) { |
| 281 | a.Effects["x"] = Effect{ID: "x", OperationID: "other", AppID: "app", Resource: "other", Action: "test", State: "prepared"} |
| 282 | }}, |
| 283 | {"registration", "test", func(a *AppAdmission) { a.Registrations["r"] = Registration{} }}, |
| 284 | {"restore", "restore", func(*AppAdmission) {}}, |
| 285 | {"mail", "mail", func(*AppAdmission) {}}, |
| 286 | } { |
| 287 | t.Run(tc.name, func(t *testing.T) { |
| 288 | f := newMutationFixture(t) |
| 289 | a := openAdmission("app") |
| 290 | a.State = "closed" |
| 291 | a.Op = &AppOperation{ID: "op", Kind: OpAppMutation, AppID: "app", Intent: map[string]string{"action": tc.action, MutationActionStarted: old, "gate": "g"}} |
| 292 | tc.mutate(&a) |
| 293 | seedAdmission(t, f.raw, "bab", "demo", a) |
| 294 | before, _ := ReadAdmission(t.Context(), f.raw, "bab", "demo", "app") |
| 295 | // Restore and mail keep their specialized RecoverMutation |
| 296 | // branches; the generic transition itself refuses them. |
| 297 | tryRecover := func() error { return f.w.RecoverMutation(t.Context(), "bab", "demo", "app", "op") } |
| 298 | if tc.action != "test" { |
| 299 | tryRecover = func() error { return f.w.abandonUnissuedMutation(t.Context(), "bab", "demo", "app", "op", 0) } |
| 300 | } |
| 301 | if err := tryRecover(); !errors.Is(err, ErrPending) { |
| 302 | t.Fatalf("recovered: %v", err) |
| 303 | } |
| 304 | after, _ := ReadAdmission(t.Context(), f.raw, "bab", "demo", "app") |
| 305 | if after.ETag != before.ETag { |
| 306 | t.Fatalf("refused recovery changed the admission: %+v", after.Value) |
| 307 | } |
| 308 | }) |
| 309 | } |
| 310 | } |
| 311 | |
| 312 | // Review P1-2: a definitive refusal before any write (stale etag at the |
| 313 | // boundary, validation) releases the operation; a refusal after a write |
| 314 | // keeps it. |
| 315 | func TestRunAppMutationRefusalDoesNotWedge(t *testing.T) { |
| 316 | f := newMutationFixture(t) |
| 317 | ctx := t.Context() |
| 318 | key := "apps/bab/demo/test.json" |
| 319 | err := release.WithAppMutation(ctx, f.data, "bab", "demo", "test", func(pctx context.Context) error { |
| 320 | _, err := f.data.Put(pctx, key, []byte(`{}`), "stale") |
| 321 | return err |
| 322 | }) |
| 323 | if !errors.Is(err, release.ErrCASConflict) { |
| 324 | t.Fatalf("stale etag: %v", err) |
| 325 | } |
| 326 | if a := f.admission(t); a.State != "open" || a.Op != nil || !effectsSettled(&a, "") { |
| 327 | t.Fatalf("stale-etag refusal wedged the app: %+v", a) |
| 328 | } |
| 329 | failure := errors.New("second write refused") |
| 330 | err = release.WithAppMutation(ctx, f.data, "bab", "demo", "test", func(pctx context.Context) error { |
| 331 | if _, err := f.data.Put(pctx, key, []byte(`{}`), ""); err != nil { |
| 332 | return err |
| 333 | } |
| 334 | return failure |
| 335 | }) |
| 336 | if !errors.Is(err, failure) { |
| 337 | t.Fatal(err) |
| 338 | } |
| 339 | if a := f.admission(t); a.State != "closed" || a.Op == nil { |
| 340 | t.Fatalf("an issued write released its operation: %+v", a) |
| 341 | } |
| 342 | } |
| 343 | |
amadan