| 1 | package claimactivation |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/ed25519" |
| 7 | "crypto/rand" |
| 8 | "errors" |
| 9 | "os" |
| 10 | "slices" |
| 11 | "strings" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/hostactors" |
| 16 | "github.com/carlosframework/platform/internal/hostbox" |
| 17 | "github.com/carlosframework/platform/internal/hostfloor" |
| 18 | "github.com/carlosframework/platform/internal/hostops" |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | ) |
| 22 | |
| 23 | // backfillDeployment is a dev-shaped deployment at step 8: a published |
| 24 | // roster at home region ie with a verified floor, leo in pool cohort1 with |
| 25 | // its census key bound, sophie in pool console (with or without a census |
| 26 | // key), and three instance records. |
| 27 | type backfillDeployment struct { |
| 28 | r Runner |
| 29 | keys map[string]ed25519.PrivateKey |
| 30 | records map[string]hostops.Record |
| 31 | } |
| 32 | |
| 33 | func newBackfillDeployment(t *testing.T, sophieKey bool) *backfillDeployment { |
| 34 | t.Helper() |
| 35 | return newBackfillDeploymentAt(t, sophieKey, true) |
| 36 | } |
| 37 | |
| 38 | func newBackfillDeploymentAt(t *testing.T, sophieKey, published bool) *backfillDeployment { |
| 39 | t.Helper() |
| 40 | ctx := t.Context() |
| 41 | b := &backfillDeployment{r: runnerFixture(t), keys: map[string]ed25519.PrivateKey{}, records: map[string]hostops.Record{}} |
| 42 | d := b.r.Directory |
| 43 | if err := d.Initialize(ctx, b.r.Signer); err != nil { |
| 44 | t.Fatal(err) |
| 45 | } |
| 46 | o, err := d.Read(ctx) |
| 47 | if err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | if err := hostfloor.Publish(ctx, d.Store, b.r.Signer, "bucket", hostfloor.Floor{Min: 1, MinRelease: "v1.0.0"}); err != nil { |
| 51 | t.Fatal(err) |
| 52 | } |
| 53 | f, err := hostfloor.Read(ctx, d.Store, d.PublicKey, "bucket") |
| 54 | if err != nil { |
| 55 | t.Fatal(err) |
| 56 | } |
| 57 | next := hostactors.Clone(o.Roster) |
| 58 | next.RequestLimitNS = int64(hostbox.RequestLimit) |
| 59 | next.Regions = map[string]hostactors.Region{"ie": {Account: "123456789012", AWSRegion: "eu-west-1", Namespaces: []string{"platform"}, Pools: []string{"cohort1", "console"}}} |
| 60 | next.HomeRegion = "ie" |
| 61 | if published { |
| 62 | next.Publication = "published" |
| 63 | next.FloorSHA256 = f.Digest |
| 64 | } |
| 65 | for i, box := range []struct{ ref, pool string }{{"leo-climbing", "cohort1"}, {"sophie-knitting", "console"}} { |
| 66 | row := hostactors.Box{Enrollment: strings.Repeat(string(rune('a'+i)), 32), Namespace: "platform", BoxID: box.ref, Label: box.ref, Region: "ie", Pool: box.pool, |
| 67 | Cloud: hostactors.Cloud{Provider: "aws", Account: "123456789012", Region: "eu-west-1", InstanceID: "i-" + box.ref}, EvidenceSHA256: strings.Repeat("b", 64), State: "active", LossIDs: []string{}} |
| 68 | if box.ref == "leo-climbing" || sophieKey { |
| 69 | pub, key, _ := ed25519.GenerateKey(rand.Reader) |
| 70 | row.CensusPublicKey, b.keys[box.ref] = pub, key |
| 71 | } |
| 72 | next.Boxes[box.ref] = row |
| 73 | } |
| 74 | if _, err := d.Replace(ctx, b.r.Signer, o, next); err != nil { |
| 75 | t.Fatal(err) |
| 76 | } |
| 77 | for _, rec := range []provision.Record{ |
| 78 | {Schema: 1, Account: "bab", App: "claimtrial", Host: "claimtrial.example.com", Slug: "claimtrial", Channel: "edge", Pool: "cohort1", Acquisition: "acq-claimtrial"}, |
| 79 | {Schema: 1, Account: "bab", App: "rollcheck", Host: "rollcheck.example.com", Slug: "rollcheck", Channel: "edge", Acquisition: "acq-rollcheck"}, |
| 80 | {Schema: 1, Account: "bab", App: "wakecheck", Host: "wakecheck.example.com", Slug: "wakecheck", Channel: "edge", Acquisition: "acq-wakecheck"}, |
| 81 | } { |
| 82 | b.record(t, rec) |
| 83 | } |
| 84 | return b |
| 85 | } |
| 86 | |
| 87 | func (b *backfillDeployment) record(t *testing.T, rec provision.Record) { |
| 88 | t.Helper() |
| 89 | r, err := hostops.PrepareInstance(rec, time.Date(2026, 10, 18, 9, 0, 0, 0, time.UTC), "ie") |
| 90 | if err != nil { |
| 91 | t.Fatal(err) |
| 92 | } |
| 93 | if _, err := b.r.Directory.Store.Put(t.Context(), r.Holder.Key, r.Body, ""); err != nil { |
| 94 | t.Fatal(err) |
| 95 | } |
| 96 | if r, err = hostops.ReadRecord(t.Context(), b.r.Directory.Store, r.Holder.Key); err != nil { |
| 97 | t.Fatal(err) |
| 98 | } |
| 99 | b.records[rec.Host] = r |
| 100 | } |
| 101 | |
| 102 | func (b *backfillDeployment) revision(t *testing.T) uint64 { |
| 103 | t.Helper() |
| 104 | o, err := b.r.Directory.Read(t.Context()) |
| 105 | if err != nil { |
| 106 | t.Fatal(err) |
| 107 | } |
| 108 | return o.Roster.Revision |
| 109 | } |
| 110 | |
| 111 | func (b *backfillDeployment) plan(t *testing.T, phase string) Plan { |
| 112 | return Plan{Phase: phase, Bucket: "bucket", ExpectedRevision: b.revision(t), Backfill: &BackfillInput{ID: "cohort1-20261021"}} |
| 113 | } |
| 114 | |
| 115 | // report publishes box's signed census report for the reservation, naming |
| 116 | // hosts as its route installations: what journal-migrate's census publishes. |
| 117 | func (b *backfillDeployment) report(t *testing.T, box string, hosts ...string) { |
| 118 | t.Helper() |
| 119 | o, err := b.r.Directory.Read(t.Context()) |
| 120 | if err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | report := hostops.CensusReport{Box: box, Enrollment: o.Roster.Boxes[box].Enrollment, BootID: "boot", Reservation: "cohort1-20261021", DirectoryRevision: o.Roster.Revision, |
| 124 | CapturedAt: time.Now().UTC(), Complete: true, Installations: []hostops.CensusInstallation{}, Seeds: &hostops.CensusSeeds{Journaled: []string{}, Gated: []string{}}} |
| 125 | for _, host := range hosts { |
| 126 | r := b.records[host] |
| 127 | inc, err := r.Incarnation() |
| 128 | if err != nil { |
| 129 | t.Fatal(err) |
| 130 | } |
| 131 | report.Installations = append(report.Installations, hostops.CensusInstallation{Host: host, Account: r.Holder.Account, App: r.Holder.App, Family: r.Holder.Family, Key: r.Holder.Key, Incarnation: inc, Route: true}) |
| 132 | } |
| 133 | if err := hostops.PublishCensusReport(t.Context(), b.r.Directory.Store, b.r.Directory, report, b.keys[box]); err != nil { |
| 134 | t.Fatal(err) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | func readPlan(t *testing.T, r Runner) hostops.BackfillPlan { |
| 139 | t.Helper() |
| 140 | p, _, err := readSignedPlan(t.Context(), r.Directory) |
| 141 | if err != nil { |
| 142 | t.Fatal(err) |
| 143 | } |
| 144 | return p |
| 145 | } |
| 146 | |
| 147 | // backfill-collect freezes every record the bucket lists, create-only, and |
| 148 | // prints the plan on a dry run without writing it. backfill-adopt, once each |
| 149 | // roster box has published its census, advances it under If-Match, takes |
| 150 | // the boxes from the census, and leaves unreported hosts without one. |
| 151 | func TestBackfillPhases(t *testing.T) { |
| 152 | ctx := t.Context() |
| 153 | b := newBackfillDeployment(t, true) |
| 154 | dry, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), true) |
| 155 | if err != nil || len(dry.Writes) != 1 || dry.Writes[0].Key != hostops.BackfillKey || dry.Writes[0].BeforeETag != "" || dry.Backfill == nil || len(dry.Backfill.Hosts) != 3 { |
| 156 | t.Fatalf("collect dry run = %+v, %v", dry, err) |
| 157 | } |
| 158 | if _, _, err := b.r.Directory.Store.Get(ctx, hostops.BackfillKey); !errors.Is(err, os.ErrNotExist) { |
| 159 | t.Fatal("dry run wrote the plan", err) |
| 160 | } |
| 161 | got, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false) |
| 162 | if err != nil || got.Backfill.Phase != "collecting" || len(got.Backfill.Hosts) != 3 { |
| 163 | t.Fatalf("collect = %+v, %v", got, err) |
| 164 | } |
| 165 | p := readPlan(t, b.r) |
| 166 | for host, r := range b.records { |
| 167 | h := p.Hosts[host] |
| 168 | if h.Key != r.Holder.Key || h.ETag != r.ETag || h.Acquisition != r.Holder.Acquisition || h.Box != "" || h.Region != "" { |
| 169 | t.Fatalf("frozen %s = %+v", host, h) |
| 170 | } |
| 171 | } |
| 172 | if p.Phase != "collecting" || p.DirectoryRevision != b.revision(t) || p.ID != "cohort1-20261021" { |
| 173 | t.Fatalf("plan = %+v", p) |
| 174 | } |
| 175 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); !errors.Is(err, ErrBackfillExists) { |
| 176 | t.Fatalf("second collect = %v", err) |
| 177 | } |
| 178 | // Adopt waits for every roster box's census report. |
| 179 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); err == nil { |
| 180 | t.Fatal("adopted without census reports") |
| 181 | } |
| 182 | b.report(t, "leo-climbing", "claimtrial.example.com") |
| 183 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); err == nil { |
| 184 | t.Fatal("adopted without sophie's report") |
| 185 | } |
| 186 | b.report(t, "sophie-knitting", "wakecheck.example.com") |
| 187 | wrong := b.plan(t, "backfill-adopt") |
| 188 | wrong.Backfill.ID = "other" |
| 189 | if _, err := b.r.Run(ctx, wrong, false); !errors.Is(err, ErrBackfillPlan) { |
| 190 | t.Fatalf("other reservation = %v", err) |
| 191 | } |
| 192 | dry, err = b.r.Run(ctx, b.plan(t, "backfill-adopt"), true) |
| 193 | if err != nil || len(dry.Writes) != 1 || dry.Backfill.Phase != "adopting" { |
| 194 | t.Fatalf("adopt dry run = %+v, %v", dry, err) |
| 195 | } |
| 196 | if readPlan(t, b.r).Phase != "collecting" { |
| 197 | t.Fatal("dry run advanced the plan") |
| 198 | } |
| 199 | got, err = b.r.Run(ctx, b.plan(t, "backfill-adopt"), false) |
| 200 | if err != nil || got.Backfill.Phase != "adopting" || len(got.Backfill.Reports) != 2 || len(got.Backfill.Unowned) != 1 || got.Backfill.Unowned[0] != "rollcheck.example.com" { |
| 201 | t.Fatalf("adopt = %+v, %v", got.Backfill, err) |
| 202 | } |
| 203 | if got.Writes[0].BeforeETag == "" { |
| 204 | t.Fatal("adopt wrote without If-Match") |
| 205 | } |
| 206 | p = readPlan(t, b.r) |
| 207 | if h := p.Hosts["claimtrial.example.com"]; h.Box != "leo-climbing" || h.Region != "ie" { |
| 208 | t.Fatalf("claimtrial = %+v", h) |
| 209 | } |
| 210 | if h := p.Hosts["wakecheck.example.com"]; h.Box != "sophie-knitting" { |
| 211 | t.Fatalf("wakecheck = %+v", h) |
| 212 | } |
| 213 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); !errors.Is(err, ErrBackfillPlan) { |
| 214 | t.Fatalf("second adopt = %v", err) |
| 215 | } |
| 216 | // done is only ever from collecting. |
| 217 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillPlan) { |
| 218 | t.Fatalf("done over an adopting plan = %v", err) |
| 219 | } |
| 220 | |
| 221 | // backfill-done, the empty cohort (claim trial Option B): records are |
| 222 | // deleted before the window (all of them, as the trial does; or only |
| 223 | // leo's, the two-host shape), sophie has no census key, leo reports |
| 224 | // nothing. done writes done under If-Match and lifts the create freeze; |
| 225 | // a dry run writes nothing; a second done is refused. |
| 226 | for name, keep := range map[string][]string{"zero hosts": nil, "two hosts": {"rollcheck.example.com", "wakecheck.example.com"}} { |
| 227 | t.Run(name, func(t *testing.T) { |
| 228 | ctx := t.Context() |
| 229 | e := newBackfillDeployment(t, false) |
| 230 | e.deleteRecordsExcept(t, keep...) |
| 231 | dry, err := e.r.Run(ctx, e.plan(t, "backfill-collect"), true) |
| 232 | if err != nil || len(dry.Writes) != 1 || len(dry.Backfill.Hosts) != len(keep) { |
| 233 | t.Fatalf("collect dry run = %+v, %v", dry, err) |
| 234 | } |
| 235 | if len(keep) == 0 && !bytes.Contains(dry.Writes[0].Payload, []byte(`"hosts":{}`)) { |
| 236 | t.Fatalf("zero-host plan payload = %s; want \"hosts\":{}", dry.Writes[0].Payload) |
| 237 | } |
| 238 | if _, err := e.r.Run(ctx, e.plan(t, "backfill-collect"), false); err != nil { |
| 239 | t.Fatal(err) |
| 240 | } |
| 241 | collecting := readPlan(t, e.r) |
| 242 | if collecting.Hosts == nil || len(collecting.Hosts) != len(keep) { |
| 243 | t.Fatalf("collecting hosts = %#v", collecting.Hosts) |
| 244 | } |
| 245 | _, collectingTag, err := e.r.Directory.Store.Get(ctx, hostops.BackfillKey) |
| 246 | if err != nil { |
| 247 | t.Fatal(err) |
| 248 | } |
| 249 | // Leo has not reported yet. |
| 250 | if _, err := e.r.Run(ctx, e.plan(t, "backfill-done"), false); !errors.Is(err, os.ErrNotExist) { |
| 251 | t.Fatalf("done before leo's report = %v", err) |
| 252 | } |
| 253 | e.report(t, "leo-climbing") |
| 254 | wrong := e.plan(t, "backfill-done") |
| 255 | wrong.Backfill.ID = "other" |
| 256 | if _, err := e.r.Run(ctx, wrong, false); !errors.Is(err, ErrBackfillPlan) { |
| 257 | t.Fatalf("done for another reservation = %v", err) |
| 258 | } |
| 259 | dry, err = e.r.Run(ctx, e.plan(t, "backfill-done"), true) |
| 260 | if err != nil || len(dry.Writes) != 1 || dry.Writes[0].Key != hostops.BackfillKey || dry.Writes[0].BeforeETag != collectingTag || dry.Backfill.Phase != "done" { |
| 261 | t.Fatalf("done dry run = %+v, %v", dry, err) |
| 262 | } |
| 263 | if readPlan(t, e.r).Phase != "collecting" { |
| 264 | t.Fatal("dry run moved the plan") |
| 265 | } |
| 266 | if err := hostops.CheckCreateFreeze(ctx, e.r.Directory.Store, e.r.Directory.PublicKey); !errors.Is(err, hostops.ErrPending) { |
| 267 | t.Fatalf("freeze after a dry run = %v", err) |
| 268 | } |
| 269 | got, err := e.r.Run(ctx, e.plan(t, "backfill-done"), false) |
| 270 | if err != nil || got.Backfill.Phase != "done" || len(got.Backfill.Hosts) != len(keep) || len(got.Backfill.Reports) != 1 || got.Backfill.Reports["leo-climbing"] == "" || |
| 271 | len(got.Backfill.Unreported) != 1 || got.Backfill.Unreported[0] != "sophie-knitting" || len(got.Backfill.Unowned) != 0 { |
| 272 | t.Fatalf("done = %+v, %v", got.Backfill, err) |
| 273 | } |
| 274 | if len(got.Writes) != 1 || got.Writes[0].BeforeETag != collectingTag { |
| 275 | t.Fatalf("done wrote %+v; want one If-Match write over the collecting plan", got.Writes) |
| 276 | } |
| 277 | done := readPlan(t, e.r) |
| 278 | if done.Phase != "done" || done.Hosts == nil || len(done.Hosts) != len(keep) { |
| 279 | t.Fatalf("done plan = %#v", done) |
| 280 | } |
| 281 | if err := hostops.CheckCreateFreeze(ctx, e.r.Directory.Store, e.r.Directory.PublicKey); err != nil { |
| 282 | t.Fatalf("freeze after done = %v", err) |
| 283 | } |
| 284 | if _, err := e.r.Run(ctx, e.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillPlan) { |
| 285 | t.Fatalf("second done = %v", err) |
| 286 | } |
| 287 | }) |
| 288 | } |
| 289 | } |
| 290 | |
| 291 | func (b *backfillDeployment) deleteRecordsExcept(t *testing.T, keep ...string) { |
| 292 | t.Helper() |
| 293 | for host, r := range b.records { |
| 294 | if slices.Contains(keep, host) { |
| 295 | continue |
| 296 | } |
| 297 | if err := release.DeleteIfMatch(t.Context(), b.r.Directory.Store, r.Holder.Key, r.ETag); err != nil { |
| 298 | t.Fatal(err) |
| 299 | } |
| 300 | delete(b.records, host) |
| 301 | } |
| 302 | } |
| 303 | |
| 304 | // backfill-done is fail-closed on installations, with no override: leo's |
| 305 | // report still lists claimtrial (B1 skipped), so done refuses with the |
| 306 | // empty-cohort refusal and the plan stays collecting. A record that changed |
| 307 | // since collect refuses with the records refusal. |
| 308 | func TestBackfillDoneRefusesAPopulatedCohort(t *testing.T) { |
| 309 | ctx := t.Context() |
| 310 | b := newBackfillDeployment(t, false) |
| 311 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); err != nil { |
| 312 | t.Fatal(err) |
| 313 | } |
| 314 | b.report(t, "leo-climbing", "claimtrial.example.com") |
| 315 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-done"), true); !errors.Is(err, ErrBackfillInstallations) || !errors.Is(err, hostops.ErrDoneInstallations) { |
| 316 | t.Fatalf("done dry run over an installation = %v", err) |
| 317 | } |
| 318 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillInstallations) { |
| 319 | t.Fatalf("done over an installation = %v", err) |
| 320 | } |
| 321 | if readPlan(t, b.r).Phase != "collecting" { |
| 322 | t.Fatal("plan moved") |
| 323 | } |
| 324 | |
| 325 | m := newBackfillDeployment(t, false) |
| 326 | if _, err := m.r.Run(ctx, m.plan(t, "backfill-collect"), false); err != nil { |
| 327 | t.Fatal(err) |
| 328 | } |
| 329 | m.report(t, "leo-climbing") |
| 330 | r := m.records["wakecheck.example.com"] |
| 331 | if _, err := m.r.Directory.Store.Put(ctx, r.Holder.Key, append(append([]byte(nil), r.Body...), '\n'), r.ETag); err != nil { |
| 332 | t.Fatal(err) |
| 333 | } |
| 334 | if _, err := m.r.Run(ctx, m.plan(t, "backfill-done"), false); !errors.Is(err, ErrBackfillRecords) { |
| 335 | t.Fatalf("done over a changed record = %v", err) |
| 336 | } |
| 337 | if readPlan(t, m.r).Phase != "collecting" { |
| 338 | t.Fatal("plan moved") |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | // The dev roster at step 8: sophie is a roster box with no census key, so |
| 343 | // no census report of hers can verify, and backfill-adopt refuses. The plan |
| 344 | // stays collecting. |
| 345 | func TestBackfillAdoptRefusesABoxWithoutCensusKey(t *testing.T) { |
| 346 | ctx := t.Context() |
| 347 | b := newBackfillDeployment(t, false) |
| 348 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); err != nil { |
| 349 | t.Fatal(err) |
| 350 | } |
| 351 | b.report(t, "leo-climbing", "claimtrial.example.com") |
| 352 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-adopt"), false); err == nil { |
| 353 | t.Fatal("adopted with a roster box that cannot report") |
| 354 | } |
| 355 | if readPlan(t, b.r).Phase != "collecting" { |
| 356 | t.Fatal("plan moved") |
| 357 | } |
| 358 | } |
| 359 | |
| 360 | // Refusals up front: a roster that is not published, or has no home region |
| 361 | // or no verified floor; a reservation id that is not one safe segment; a |
| 362 | // record with no acquisition. |
| 363 | func TestBackfillCollectRefusesUpFront(t *testing.T) { |
| 364 | ctx := t.Context() |
| 365 | if _, err := newBackfillDeploymentAt(t, true, false).r.Run(ctx, Plan{Phase: "backfill-collect", Bucket: "bucket", ExpectedRevision: 2, Backfill: &BackfillInput{ID: "cohort1-20261021"}}, true); !errors.Is(err, ErrBackfillState) { |
| 366 | t.Fatalf("collect on an unpublished roster = %v", err) |
| 367 | } |
| 368 | gone := newBackfillDeployment(t, true) |
| 369 | for _, key := range []string{hostfloor.Key, hostfloor.AuthKey} { |
| 370 | if err := gone.r.Directory.Store.Delete(ctx, key); err != nil { |
| 371 | t.Fatal(err) |
| 372 | } |
| 373 | } |
| 374 | if _, err := gone.r.Run(ctx, gone.plan(t, "backfill-collect"), true); !errors.Is(err, ErrBackfillState) { |
| 375 | t.Fatalf("collect with no verified floor = %v", err) |
| 376 | } |
| 377 | b := newBackfillDeployment(t, true) |
| 378 | for _, id := range []string{"", "a/b", "..", "UPPER", strings.Repeat("a", 65)} { |
| 379 | p := b.plan(t, "backfill-collect") |
| 380 | p.Backfill.ID = id |
| 381 | if _, err := b.r.Run(ctx, p, true); !errors.Is(err, ErrActivation) { |
| 382 | t.Fatalf("id %q = %v", id, err) |
| 383 | } |
| 384 | } |
| 385 | p := b.plan(t, "backfill-collect") |
| 386 | p.Backfill = nil |
| 387 | if _, err := b.r.Run(ctx, p, true); !errors.Is(err, ErrActivation) { |
| 388 | t.Fatalf("no backfill input = %v", err) |
| 389 | } |
| 390 | b.record(t, provision.Record{Schema: 1, Account: "bab", App: "legacy", Host: "legacy.example.com", Slug: "legacy", Channel: "edge"}) |
| 391 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), true); !errors.Is(err, ErrBackfillRecords) { |
| 392 | t.Fatalf("record without acquisition = %v", err) |
| 393 | } |
| 394 | } |
| 395 | |
| 396 | // A record that moves while the plan is being written leaves a plan the box |
| 397 | // census would refuse; collect reports it rather than succeeding. |
| 398 | type movingRecord struct { |
| 399 | release.Store |
| 400 | key string |
| 401 | body []byte |
| 402 | done bool |
| 403 | } |
| 404 | |
| 405 | func (s *movingRecord) Put(ctx context.Context, key string, raw []byte, before string) (string, error) { |
| 406 | tag, err := s.Store.Put(ctx, key, raw, before) |
| 407 | if err == nil && key == hostops.BackfillKey && !s.done { |
| 408 | s.done = true |
| 409 | _, old, _ := s.Store.Get(ctx, s.key) |
| 410 | if _, e := s.Store.Put(ctx, s.key, s.body, old); e != nil { |
| 411 | return "", e |
| 412 | } |
| 413 | } |
| 414 | return tag, err |
| 415 | } |
| 416 | |
| 417 | func TestBackfillCollectRefusesARecordThatMoved(t *testing.T) { |
| 418 | b := newBackfillDeployment(t, true) |
| 419 | r := b.records["claimtrial.example.com"] |
| 420 | body := append([]byte(nil), r.Body...) |
| 421 | b.r.Directory.Store = &movingRecord{Store: b.r.Directory.Store, key: r.Holder.Key, body: append(body, '\n')} |
| 422 | if _, err := b.r.Run(t.Context(), b.plan(t, "backfill-collect"), false); !errors.Is(err, ErrBackfillRecords) { |
| 423 | t.Fatalf("collect over a moving record = %v", err) |
| 424 | } |
| 425 | } |
| 426 | |
| 427 | // windowStore runs afterGet once on the first Get of getKey and afterPut |
| 428 | // once after the first Put of putKey: a console record writer racing the |
| 429 | // claims run's backfill-done. |
| 430 | type windowStore struct { |
| 431 | release.Store |
| 432 | getKey, putKey string |
| 433 | afterGet, afterPut func() |
| 434 | } |
| 435 | |
| 436 | func (s *windowStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 437 | b, tag, err := s.Store.Get(ctx, key) |
| 438 | if err == nil && key == s.getKey && s.afterGet != nil { |
| 439 | f := s.afterGet |
| 440 | s.afterGet = nil |
| 441 | f() |
| 442 | } |
| 443 | return b, tag, err |
| 444 | } |
| 445 | |
| 446 | func (s *windowStore) Put(ctx context.Context, key string, b []byte, tag string) (string, error) { |
| 447 | got, err := s.Store.Put(ctx, key, b, tag) |
| 448 | if err == nil && key == s.putKey && s.afterPut != nil { |
| 449 | f := s.afterPut |
| 450 | s.afterPut = nil |
| 451 | f() |
| 452 | } |
| 453 | return got, err |
| 454 | } |
| 455 | |
| 456 | // A record created during the window reaches the operator as the records |
| 457 | // refusal: while leo's report is read, done refuses before writing and the |
| 458 | // plan stays collecting; after done's write, the refusal carries |
| 459 | // ErrDoneRecordsChanged and the plan is done. |
| 460 | func TestBackfillDoneRecordCreatedDuringTheWindow(t *testing.T) { |
| 461 | for _, after := range []bool{false, true} { |
| 462 | name := map[bool]string{false: "during report retrieval", true: "after the write"}[after] |
| 463 | t.Run(name, func(t *testing.T) { |
| 464 | ctx := t.Context() |
| 465 | b := newBackfillDeployment(t, false) |
| 466 | if _, err := b.r.Run(ctx, b.plan(t, "backfill-collect"), false); err != nil { |
| 467 | t.Fatal(err) |
| 468 | } |
| 469 | b.report(t, "leo-climbing") |
| 470 | fired := false |
| 471 | create := func() { |
| 472 | fired = true |
| 473 | b.record(t, provision.Record{Schema: 1, Account: "bab", App: "late", Host: "late.example.com", Slug: "late", Channel: "edge", Acquisition: "acq-late"}) |
| 474 | } |
| 475 | s := &windowStore{Store: b.r.Directory.Store} |
| 476 | if after { |
| 477 | s.putKey, s.afterPut = hostops.BackfillKey, create |
| 478 | } else { |
| 479 | s.getKey, s.afterGet = "control/host-census/cohort1-20261021/leo-climbing.json", create |
| 480 | } |
| 481 | b.r.Directory.Store = s |
| 482 | _, err := b.r.Run(ctx, b.plan(t, "backfill-done"), false) |
| 483 | if !fired { |
| 484 | t.Fatal("the window's hook never ran") |
| 485 | } |
| 486 | if !errors.Is(err, ErrBackfillRecords) || errors.Is(err, hostops.ErrDoneRecordsChanged) != after { |
| 487 | t.Fatalf("done = %v", err) |
| 488 | } |
| 489 | if want := map[bool]string{false: "collecting", true: "done"}[after]; readPlan(t, b.r).Phase != want { |
| 490 | t.Fatalf("plan phase = %s; want %s", readPlan(t, b.r).Phase, want) |
| 491 | } |
| 492 | }) |
| 493 | } |
| 494 | } |
| 495 | |
amadan