| 1 | package objstore |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "strings" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | // queueFixture is a provisioned, scanning-enabled store with an armed engine |
| 15 | // and n objects the scanner knows about. |
| 16 | func queueFixture(t *testing.T, n int) (release.Store, *MemScanner, *QueueIndex, string) { |
| 17 | t.Helper() |
| 18 | s := release.NewDirStore(t.TempDir()) |
| 19 | _, pub := testSealKeypair(t) |
| 20 | p := NewMemProvisioner("eu-west-1") |
| 21 | rec := rotateFixture(t, s, pub, p) |
| 22 | enableScanning(t, s, "bab", "titogo") |
| 23 | |
| 24 | sc := NewMemScanner() |
| 25 | bucket := rec.Provisioned.Bucket |
| 26 | for i := 0; i < n; i++ { |
| 27 | sc.PutObject(bucket, objectName(i), "etag-"+objectName(i), time.Now().UTC(), 10) |
| 28 | } |
| 29 | return s, sc, NewQueueIndex(), bucket |
| 30 | } |
| 31 | |
| 32 | func enqueueN(t *testing.T, s release.Store, idx *QueueIndex, n int) { |
| 33 | t.Helper() |
| 34 | for i := 0; i < n; i++ { |
| 35 | if _, err := Enqueue(context.Background(), s, idx, "bab", "titogo", &QueueItem{ |
| 36 | Key: objectName(i), By: "app", |
| 37 | }, time.Now()); err != nil { |
| 38 | t.Fatal(err) |
| 39 | } |
| 40 | } |
| 41 | } |
| 42 | |
| 43 | // Ids sort by time, so a prefix listing IS the queue in order. |
| 44 | func TestQueueIDSortsByTime(t *testing.T) { |
| 45 | base := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC) |
| 46 | var prev string |
| 47 | for i := 0; i < 200; i++ { |
| 48 | id := NewQueueID(base.Add(time.Duration(i) * time.Millisecond)) |
| 49 | if id <= prev { |
| 50 | t.Fatalf("id %q does not sort after %q", id, prev) |
| 51 | } |
| 52 | prev = id |
| 53 | } |
| 54 | // Two in the same millisecond must not collide. |
| 55 | seen := map[string]bool{} |
| 56 | for i := 0; i < 1000; i++ { |
| 57 | id := NewQueueID(base) |
| 58 | if seen[id] { |
| 59 | t.Fatal("two ids in the same millisecond collided") |
| 60 | } |
| 61 | seen[id] = true |
| 62 | } |
| 63 | } |
| 64 | |
| 65 | // The object first, the index second. A crash between the two strands one item |
| 66 | // until the boot rebuild; the other order would mark an app whose item was |
| 67 | // never written, and a 202 sent in between would stop the client retrying. |
| 68 | // It is the ORDER that is under test, so the assertions have to be able to see |
| 69 | // it. Asserting only on the end state passes just as happily with the two |
| 70 | // swapped, which is the mutation that matters. |
| 71 | func TestEnqueueWritesObjectBeforeBump(t *testing.T) { |
| 72 | ctx := context.Background() |
| 73 | s, _, idx, _ := queueFixture(t, 1) |
| 74 | |
| 75 | // (a) At the instant the object is being written, the app is not yet |
| 76 | // marked. Bump-first fails here. |
| 77 | var markedDuringWrite int |
| 78 | watching := &queueStore{Store: s, onPut: func(string) { markedDuringWrite = idx.Len() }} |
| 79 | id, err := Enqueue(ctx, watching, idx, "bab", "titogo", &QueueItem{ |
| 80 | Key: "uploads/a", By: "app", |
| 81 | }, time.Now()) |
| 82 | if err != nil { |
| 83 | t.Fatal(err) |
| 84 | } |
| 85 | if markedDuringWrite != 0 { |
| 86 | t.Fatal("the index was bumped before the item object was written") |
| 87 | } |
| 88 | if _, _, err := s.Get(ctx, QueueKey("bab", "titogo", id)); err != nil { |
| 89 | t.Fatalf("item object was not written: %v", err) |
| 90 | } |
| 91 | if idx.Len() != 1 { |
| 92 | t.Fatal("the app was not marked") |
| 93 | } |
| 94 | |
| 95 | // (b) And a write that FAILS leaves the app unmarked, so nothing ever |
| 96 | // points at an item that does not exist. |
| 97 | fresh := NewQueueIndex() |
| 98 | failing := &queueStore{Store: s, putErr: errors.New("the bucket said no")} |
| 99 | if _, err := Enqueue(ctx, failing, fresh, "bab", "titogo", &QueueItem{ |
| 100 | Key: "uploads/b", By: "app", |
| 101 | }, time.Now()); err == nil { |
| 102 | t.Fatal("a failed write reported success") |
| 103 | } |
| 104 | if fresh.Len() != 0 { |
| 105 | t.Fatal("the app was marked for an item that was never written") |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | // THE test this design needed. The drain caps work per tick, so on a flooded |
| 110 | // app it deliberately leaves items behind with no enqueue to bump anything — |
| 111 | // and an earlier draft released the entry whenever the generation was |
| 112 | // unchanged, deleting the pointer to exactly those items. They would then sit |
| 113 | // until the next console boot, and the callback and long-poll wake would be |
| 114 | // lost for the app the cap exists for. |
| 115 | func TestIndexEntrySurvivesPartialDrain(t *testing.T) { |
| 116 | ctx := context.Background() |
| 117 | s, sc, idx, _ := queueFixture(t, 5) |
| 118 | enqueueN(t, s, idx, 5) |
| 119 | |
| 120 | done, listedEmpty, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 2, time.Now()) |
| 121 | if err != nil { |
| 122 | t.Fatal(err) |
| 123 | } |
| 124 | if len(done) != 2 { |
| 125 | t.Fatalf("drained %d, want the cap of 2", len(done)) |
| 126 | } |
| 127 | if listedEmpty { |
| 128 | t.Fatal("a capped listing reported itself empty") |
| 129 | } |
| 130 | if idx.Len() != 1 { |
| 131 | t.Fatal("the entry pointing at the remaining items was released") |
| 132 | } |
| 133 | |
| 134 | // Successive calls finish the job. |
| 135 | for i := 0; i < 5 && idx.Len() > 0; i++ { |
| 136 | if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 2, time.Now()); err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | } |
| 140 | if idx.Len() != 0 { |
| 141 | t.Fatal("successive drains never emptied the queue") |
| 142 | } |
| 143 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 144 | if err != nil || len(left) != 0 { |
| 145 | t.Fatalf("items left behind: %v, %v", left, err) |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | // An enqueue landing mid-drain must not have its entry released underneath it. |
| 150 | // |
| 151 | // Driven through DRAINAPP, not through Observe and Release by hand: what the |
| 152 | // generation counter protects is an ORDERING inside DrainApp — Observe before |
| 153 | // the listing — and a test that calls the three methods itself proves the |
| 154 | // methods work while leaving the ordering free to be wrong. Moving Observe |
| 155 | // below the listing passes such a test and fails this one. |
| 156 | func TestIndexEntrySurvivesEnqueueDuringDrain(t *testing.T) { |
| 157 | ctx := context.Background() |
| 158 | s, sc, idx, _ := queueFixture(t, 2) |
| 159 | enqueueN(t, s, idx, 1) |
| 160 | |
| 161 | // Drain the one item. The entry stays, because a consumed page may have |
| 162 | // more behind it. |
| 163 | if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err != nil { |
| 164 | t.Fatal(err) |
| 165 | } |
| 166 | if idx.Len() != 1 { |
| 167 | t.Fatal("a consumed page released the entry before an empty listing confirmed it") |
| 168 | } |
| 169 | |
| 170 | // Now the queue really is empty — except that an enqueue lands after the |
| 171 | // listing has been taken and before the release decision. |
| 172 | racing := &queueStore{Store: s} |
| 173 | racing.afterList = func() { |
| 174 | if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{ |
| 175 | Key: objectName(1), By: "app", |
| 176 | }, time.Now()); err != nil { |
| 177 | t.Error(err) |
| 178 | } |
| 179 | } |
| 180 | if _, listedEmpty, err := DrainApp(ctx, racing, sc, idx, "bab", "titogo", 10, time.Now()); err != nil || !listedEmpty { |
| 181 | t.Fatalf("drain = %v, %v", listedEmpty, err) |
| 182 | } |
| 183 | if idx.Len() != 1 { |
| 184 | t.Fatal("the entry for an item enqueued during the drain was released — nothing points at it until boot") |
| 185 | } |
| 186 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 187 | if err != nil || len(left) != 1 { |
| 188 | t.Fatalf("the late item = %v, %v", left, err) |
| 189 | } |
| 190 | } |
| 191 | |
| 192 | // Release needs BOTH conditions. Generation alone is the bug above. |
| 193 | func TestIndexReleasedOnlyOnEmptyListing(t *testing.T) { |
| 194 | ref := AppRef{Account: "bab", App: "titogo"} |
| 195 | idx := NewQueueIndex() |
| 196 | idx.Bump(ref) |
| 197 | observed := idx.Observe(ref) |
| 198 | |
| 199 | if idx.Release(ref, observed, false) { |
| 200 | t.Fatal("released on a non-empty listing") |
| 201 | } |
| 202 | if !idx.Release(ref, observed, true) { |
| 203 | t.Fatal("did not release on an empty listing with an unchanged generation") |
| 204 | } |
| 205 | } |
| 206 | |
| 207 | // The consent chain, re-checked at the drain because the record can change |
| 208 | // between enqueue and drain and THIS is where bytes are read. |
| 209 | func TestDrainAppRefusesAndDeletesWhenNotPermitted(t *testing.T) { |
| 210 | ctx := context.Background() |
| 211 | |
| 212 | for name, break_ := range map[string]func(t *testing.T, s release.Store){ |
| 213 | "no attestation": func(t *testing.T, s release.Store) { |
| 214 | rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| 215 | rec.Scanning.Requested = nil |
| 216 | if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| 217 | t.Fatal(err) |
| 218 | } |
| 219 | }, |
| 220 | "attested encrypted": func(t *testing.T, s release.Store) { |
| 221 | rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| 222 | rec.Scanning.Requested.Encrypted = true |
| 223 | if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| 224 | t.Fatal(err) |
| 225 | } |
| 226 | }, |
| 227 | "engine not enabled for the store": func(t *testing.T, s release.Store) { |
| 228 | rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| 229 | rec.Scanning.Enabled = nil |
| 230 | if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| 231 | t.Fatal(err) |
| 232 | } |
| 233 | }, |
| 234 | "app in the trash": func(t *testing.T, s release.Store) { |
| 235 | if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil { |
| 236 | t.Fatal(err) |
| 237 | } |
| 238 | }, |
| 239 | } { |
| 240 | t.Run(name, func(t *testing.T) { |
| 241 | s, sc, idx, _ := queueFixture(t, 2) |
| 242 | enqueueN(t, s, idx, 2) |
| 243 | break_(t, s) |
| 244 | |
| 245 | _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 246 | if !errors.Is(err, ErrQueueNotPermitted) { |
| 247 | t.Fatalf("err = %v, want ErrQueueNotPermitted", err) |
| 248 | } |
| 249 | // The items are DELETED, not kept: none of these can change back |
| 250 | // in a way that makes them useful, and leaving them would block |
| 251 | // the app's name for thirty days in the trash case. |
| 252 | left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 253 | if lerr != nil || len(left) != 0 { |
| 254 | t.Fatalf("items survived a permanent refusal: %v, %v", left, lerr) |
| 255 | } |
| 256 | }) |
| 257 | } |
| 258 | } |
| 259 | |
| 260 | // Disabled deployment-wide is REVERSIBLE, so the items are kept. The refusals |
| 261 | // above are not, so they are deleted. Getting these two the same way round |
| 262 | // would either strand a tenant's queue or lose it. |
| 263 | func TestDrainAppKeepsItemsWhenDeploymentDisabled(t *testing.T) { |
| 264 | ctx := context.Background() |
| 265 | s, sc, idx, _ := queueFixture(t, 2) |
| 266 | enqueueN(t, s, idx, 2) |
| 267 | |
| 268 | av, etag, err := release.GetAntivirus(ctx, s) |
| 269 | if err != nil { |
| 270 | t.Fatal(err) |
| 271 | } |
| 272 | av.Enabled = false |
| 273 | if _, err := release.PutAntivirus(ctx, s, av, etag); err != nil { |
| 274 | t.Fatal(err) |
| 275 | } |
| 276 | |
| 277 | if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); !errors.Is(err, ErrScanningDisabled) { |
| 278 | t.Fatalf("err = %v, want ErrScanningDisabled", err) |
| 279 | } |
| 280 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 281 | if err != nil || len(left) != 2 { |
| 282 | t.Fatalf("a reversible refusal dropped the queue: %v, %v", left, err) |
| 283 | } |
| 284 | if idx.Len() != 1 { |
| 285 | t.Fatal("the index entry was released on a reversible refusal") |
| 286 | } |
| 287 | } |
| 288 | |
| 289 | // The drain re-issues failed and denied itself. Without this, an object the |
| 290 | // sweep last tagged failed would be answered from that tag forever and never |
| 291 | // rescanned on request — on-demand strictly less able than the hourly pass it |
| 292 | // exists to improve on. |
| 293 | func TestDrainAppReIssuesFailedAndDenied(t *testing.T) { |
| 294 | ctx := context.Background() |
| 295 | for _, stuck := range []Verdict{VerdictFailed, VerdictDenied} { |
| 296 | t.Run(string(stuck), func(t *testing.T) { |
| 297 | s, sc, idx, bucket := queueFixture(t, 1) |
| 298 | key := objectName(0) |
| 299 | // A stale verdict already on the object, at its current version. |
| 300 | _, obj, err := sc.Status(ctx, bucket, key) |
| 301 | if err != nil { |
| 302 | t.Fatal(err) |
| 303 | } |
| 304 | sc.Put(bucket, key, obj.Version(), stuck) |
| 305 | enqueueN(t, s, idx, 1) |
| 306 | |
| 307 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 308 | if err != nil { |
| 309 | t.Fatal(err) |
| 310 | } |
| 311 | if len(done) != 1 { |
| 312 | t.Fatalf("drained %d items", len(done)) |
| 313 | } |
| 314 | if !done[0].Scanned { |
| 315 | t.Fatalf("a %q verdict was answered from the tag rather than rescanned", stuck) |
| 316 | } |
| 317 | if done[0].Verdict != VerdictClean { |
| 318 | t.Fatalf("verdict = %q, want the rescan's answer", done[0].Verdict) |
| 319 | } |
| 320 | }) |
| 321 | } |
| 322 | } |
| 323 | |
| 324 | // A verdict already recorded at the current version costs a tag read and no |
| 325 | // fetch. That is the amplification guard: re-requesting a judged key must not |
| 326 | // re-stream its bytes. |
| 327 | func TestDrainAppReusesAJudgedVerdict(t *testing.T) { |
| 328 | ctx := context.Background() |
| 329 | s, sc, idx, bucket := queueFixture(t, 1) |
| 330 | key := objectName(0) |
| 331 | _, obj, err := sc.Status(ctx, bucket, key) |
| 332 | if err != nil { |
| 333 | t.Fatal(err) |
| 334 | } |
| 335 | sc.Put(bucket, key, obj.Version(), VerdictClean) |
| 336 | enqueueN(t, s, idx, 1) |
| 337 | |
| 338 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 339 | if err != nil { |
| 340 | t.Fatal(err) |
| 341 | } |
| 342 | if len(done) != 1 || done[0].Scanned { |
| 343 | t.Fatalf("a judged key was rescanned: %+v", done) |
| 344 | } |
| 345 | if done[0].Verdict != VerdictClean { |
| 346 | t.Fatalf("verdict = %q", done[0].Verdict) |
| 347 | } |
| 348 | } |
| 349 | |
| 350 | // A verdict the object cannot carry goes to a result object instead, and the |
| 351 | // outcome says so — an infected object with Recorded false is NOT covered by |
| 352 | // the enforcement deny, because the deny matches a tag that is not there. |
| 353 | func TestDrainAppWritesResultWhenUnrecorded(t *testing.T) { |
| 354 | ctx := context.Background() |
| 355 | s, sc, idx, bucket := queueFixture(t, 1) |
| 356 | key := objectName(0) |
| 357 | sc.Unrecordable(key) |
| 358 | enqueueN(t, s, idx, 1) |
| 359 | |
| 360 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 361 | if err != nil { |
| 362 | t.Fatal(err) |
| 363 | } |
| 364 | if len(done) != 1 || done[0].Recorded { |
| 365 | t.Fatalf("outcome = %+v, want recorded false", done) |
| 366 | } |
| 367 | _, obj, err := sc.Status(ctx, bucket, key) |
| 368 | if err != nil { |
| 369 | t.Fatal(err) |
| 370 | } |
| 371 | res, err := ReadScanResult(ctx, s, "bab", "titogo", key, obj.Version()) |
| 372 | if err != nil || res == nil { |
| 373 | t.Fatalf("no result object for an unrecorded verdict: %+v, %v", res, err) |
| 374 | } |
| 375 | if res.Verdict != string(VerdictClean) { |
| 376 | t.Fatalf("result verdict = %q", res.Verdict) |
| 377 | } |
| 378 | } |
| 379 | |
| 380 | // And when a later scan DOES persist a verdict, the fallback goes — it must |
| 381 | // not outlive the real answer. |
| 382 | func TestDrainAppDeletesResultWhenLaterRecorded(t *testing.T) { |
| 383 | ctx := context.Background() |
| 384 | s, sc, idx, bucket := queueFixture(t, 1) |
| 385 | key := objectName(0) |
| 386 | |
| 387 | sc.Unrecordable(key) |
| 388 | enqueueN(t, s, idx, 1) |
| 389 | if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err != nil { |
| 390 | t.Fatal(err) |
| 391 | } |
| 392 | if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", key)); err != nil { |
| 393 | t.Fatalf("setup: no result object: %v", err) |
| 394 | } |
| 395 | |
| 396 | // The tag limit clears, and the object is re-requested. |
| 397 | sc2 := NewMemScanner() |
| 398 | _, obj, _ := sc.Status(ctx, bucket, key) |
| 399 | sc2.PutObject(bucket, key, obj.ETag, obj.Modified, obj.Size) |
| 400 | if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{ |
| 401 | Key: key, By: "app", |
| 402 | }, time.Now()); err != nil { |
| 403 | t.Fatal(err) |
| 404 | } |
| 405 | if _, _, err := DrainApp(ctx, s, sc2, idx, "bab", "titogo", 10, time.Now()); err != nil { |
| 406 | t.Fatal(err) |
| 407 | } |
| 408 | if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", key)); !errors.Is(err, os.ErrNotExist) { |
| 409 | t.Fatalf("the fallback outlived the real verdict: %v", err) |
| 410 | } |
| 411 | } |
| 412 | |
| 413 | // Expired items are dropped rather than scanned: after a sweep cycle the |
| 414 | // ordinary pass has had its chance, and a stale on-demand answer is worth less |
| 415 | // than none. |
| 416 | func TestDrainAppDropsExpiredItems(t *testing.T) { |
| 417 | ctx := context.Background() |
| 418 | s, sc, idx, _ := queueFixture(t, 1) |
| 419 | if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{ |
| 420 | Key: objectName(0), By: "app", |
| 421 | }, time.Now().Add(-48*time.Hour)); err != nil { |
| 422 | t.Fatal(err) |
| 423 | } |
| 424 | |
| 425 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 426 | if err != nil { |
| 427 | t.Fatal(err) |
| 428 | } |
| 429 | if len(done) != 0 { |
| 430 | t.Fatalf("an expired item was scanned: %+v", done) |
| 431 | } |
| 432 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 433 | if err != nil || len(left) != 0 { |
| 434 | t.Fatalf("expired item not collected: %v, %v", left, err) |
| 435 | } |
| 436 | } |
| 437 | |
| 438 | // The same key twice in one tick is scanned once. This is what stops one app's |
| 439 | // flood being amplified into repeated fetches of the same bytes. |
| 440 | // |
| 441 | // BUT DEDUPING THE WORK IS NOT DEDUPING THE PROMISES. Every enqueue was |
| 442 | // answered 202, and every one that named a callback path is owed a delivery. |
| 443 | // An earlier draft deleted the duplicate items along with their paths, so two |
| 444 | // requests for one key with different callbacks got one callback and the other |
| 445 | // app endpoint waited forever — a silent loss the drain's own test blessed by |
| 446 | // asserting only that one outcome came back. |
| 447 | func TestDrainAppDedupesWorkNotCallbacks(t *testing.T) { |
| 448 | ctx := context.Background() |
| 449 | s, sc, idx, _ := queueFixture(t, 1) |
| 450 | for _, path := range []string{"/hooks/a", "/hooks/b", "/hooks/a", ""} { |
| 451 | if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{ |
| 452 | Key: objectName(0), By: "app", CallbackPath: path, |
| 453 | }, time.Now()); err != nil { |
| 454 | t.Fatal(err) |
| 455 | } |
| 456 | } |
| 457 | |
| 458 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 459 | if err != nil { |
| 460 | t.Fatal(err) |
| 461 | } |
| 462 | if len(done) != 1 { |
| 463 | t.Fatalf("drained %d outcomes for one key — the object was read more than once", len(done)) |
| 464 | } |
| 465 | if sc.Scans(objectName(0)) != 1 { |
| 466 | t.Fatalf("the engine read the object %d times for four requests", sc.Scans(objectName(0))) |
| 467 | } |
| 468 | // Both distinct paths survive; the repeat does not become a second |
| 469 | // delivery, and the request with no path adds nothing. |
| 470 | got := done[0].CallbackPaths |
| 471 | if len(got) != 2 || !containsString(got, "/hooks/a") || !containsString(got, "/hooks/b") { |
| 472 | t.Fatalf("callback paths = %v, want both distinct paths and no duplicate", got) |
| 473 | } |
| 474 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 475 | if err != nil || len(left) != 0 { |
| 476 | t.Fatalf("duplicate items left behind: %v, %v", left, err) |
| 477 | } |
| 478 | } |
| 479 | |
| 480 | // An object deleted between enqueue and drain is dropped, not an error. |
| 481 | func TestDrainAppDropsGoneObjects(t *testing.T) { |
| 482 | ctx := context.Background() |
| 483 | s, sc, idx, _ := queueFixture(t, 0) // the scanner knows about no objects |
| 484 | if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{ |
| 485 | Key: "uploads/vanished", By: "app", |
| 486 | }, time.Now()); err != nil { |
| 487 | t.Fatal(err) |
| 488 | } |
| 489 | |
| 490 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 491 | if err != nil { |
| 492 | t.Fatalf("a vanished object failed the drain: %v", err) |
| 493 | } |
| 494 | // IT IS AN OUTCOME, and this test used to assert the opposite. "There is |
| 495 | // no such object" is the honest answer to the question the app asked, and |
| 496 | // reporting nothing meant a long-poll parked on that key sat out its whole |
| 497 | // window for an answer that had already been settled — the drain deleted |
| 498 | // the item silently and woke nobody. |
| 499 | if len(done) != 1 { |
| 500 | t.Fatalf("outcomes = %+v, want the one that says the object is gone", done) |
| 501 | } |
| 502 | if !done[0].Gone { |
| 503 | t.Errorf("outcome %+v is not marked Gone, so a caller cannot tell it from a verdict", done[0]) |
| 504 | } |
| 505 | if done[0].Verdict != "" || done[0].Recorded || done[0].Scanned { |
| 506 | t.Errorf("a gone object produced a verdict-shaped outcome: %+v", done[0]) |
| 507 | } |
| 508 | left, _ := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 509 | if len(left) != 0 { |
| 510 | t.Fatal("the item for a vanished object was kept") |
| 511 | } |
| 512 | } |
| 513 | |
| 514 | // An engine that goes away keeps the items AND the entry: it is a fact about |
| 515 | // the engine, not about these objects, and the next tick should retry. |
| 516 | func TestDrainAppKeepsItemsOnEngineFailure(t *testing.T) { |
| 517 | ctx := context.Background() |
| 518 | s, sc, idx, _ := queueFixture(t, 1) |
| 519 | enqueueN(t, s, idx, 1) |
| 520 | sc.Fail("Scan", errors.New("dial unix /run/clamav/clamd.ctl: connection refused")) |
| 521 | |
| 522 | if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err == nil { |
| 523 | t.Fatal("an unreachable engine reported a successful drain") |
| 524 | } |
| 525 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 526 | if err != nil || len(left) != 1 { |
| 527 | t.Fatalf("an engine failure dropped the item: %v, %v", left, err) |
| 528 | } |
| 529 | if idx.Len() != 1 { |
| 530 | t.Fatal("an engine failure released the index entry") |
| 531 | } |
| 532 | } |
| 533 | |
| 534 | // ---- the store wrapper the ordering and error tests need ------------------- |
| 535 | |
| 536 | // queueStore wraps a store so one operation can be failed, watched or raced. |
| 537 | // |
| 538 | // The optional listings are forwarded BY HAND. An embedded interface promotes |
| 539 | // only release.Store's own methods, so a wrapper that did not forward would |
| 540 | // quietly take the drain off PageLister and the collector off ModLister — and |
| 541 | // every test using it would then exercise the fallback path instead of the one |
| 542 | // it names. |
| 543 | type queueStore struct { |
| 544 | release.Store |
| 545 | putErr error // every Put fails with this |
| 546 | onPut func(key string) // runs BEFORE the write lands |
| 547 | failGetPrefix string // reads of keys under this prefix fail |
| 548 | failGetKey string // reads of exactly this key fail |
| 549 | afterList func() // runs after the listing, before the caller sees it |
| 550 | } |
| 551 | |
| 552 | func (q *queueStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) { |
| 553 | if q.onPut != nil { |
| 554 | q.onPut(key) |
| 555 | } |
| 556 | if q.putErr != nil { |
| 557 | return "", q.putErr |
| 558 | } |
| 559 | return q.Store.Put(ctx, key, body, etag) |
| 560 | } |
| 561 | |
| 562 | func (q *queueStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 563 | if key == q.failGetKey || (q.failGetPrefix != "" && strings.HasPrefix(key, q.failGetPrefix)) { |
| 564 | return nil, "", errors.New("objstore test: the bucket had a bad moment") |
| 565 | } |
| 566 | return q.Store.Get(ctx, key) |
| 567 | } |
| 568 | |
| 569 | func (q *queueStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| 570 | keys, err := q.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit) |
| 571 | if q.afterList != nil { |
| 572 | f := q.afterList |
| 573 | q.afterList = nil // once |
| 574 | f() |
| 575 | } |
| 576 | return keys, err |
| 577 | } |
| 578 | |
| 579 | func (q *queueStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| 580 | return q.Store.(release.ModLister).ListMod(ctx, prefix) |
| 581 | } |
| 582 | |
| 583 | // Enqueue owns the timestamp, so a door cannot forget to stamp it. A zero At |
| 584 | // would mint an id sorting at year one and an item the very next drain |
| 585 | // discards as expired — a 202 that meant nothing. |
| 586 | func TestEnqueueStampsItsOwnTimestamp(t *testing.T) { |
| 587 | ctx := context.Background() |
| 588 | s, _, idx, _ := queueFixture(t, 1) |
| 589 | now := time.Date(2026, 9, 2, 9, 0, 0, 0, time.UTC) |
| 590 | |
| 591 | item := &QueueItem{Key: "uploads/a", By: "app"} // no At |
| 592 | id, err := Enqueue(ctx, s, idx, "bab", "titogo", item, now) |
| 593 | if err != nil { |
| 594 | t.Fatal(err) |
| 595 | } |
| 596 | if !item.At.Equal(now) { |
| 597 | t.Fatalf("At = %v, want the clock the caller passed", item.At) |
| 598 | } |
| 599 | if want := NewQueueID(now); id[:12] != want[:12] { |
| 600 | t.Fatalf("id %q does not carry the enqueue time (want the prefix of %q)", id, want) |
| 601 | } |
| 602 | stored, err := readQueueItem(ctx, s, QueueKey("bab", "titogo", id)) |
| 603 | if err != nil || stored.At.IsZero() { |
| 604 | t.Fatalf("stored item = %+v, %v", stored, err) |
| 605 | } |
| 606 | } |
| 607 | |
| 608 | // A HeadObject the console may not make is not a recorded verdict. |
| 609 | // |
| 610 | // Saying recorded here would put "denied" in a callback beside a claim that |
| 611 | // the object carries it, when no tag was ever read or written — and the |
| 612 | // enforcement deny matches on a tag that is not there, so the app would be |
| 613 | // told a durable answer exists that no door can produce. |
| 614 | func TestDrainAppDeniedHeadIsNotRecorded(t *testing.T) { |
| 615 | ctx := context.Background() |
| 616 | s, sc, idx, bucket := queueFixture(t, 1) |
| 617 | sc.DenyHead(bucket, objectName(0)) |
| 618 | enqueueN(t, s, idx, 1) |
| 619 | |
| 620 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 621 | if err != nil { |
| 622 | t.Fatal(err) |
| 623 | } |
| 624 | if len(done) != 1 { |
| 625 | t.Fatalf("drained %d outcomes", len(done)) |
| 626 | } |
| 627 | if done[0].Verdict != VerdictDenied { |
| 628 | t.Fatalf("verdict = %q, want denied", done[0].Verdict) |
| 629 | } |
| 630 | if done[0].Recorded { |
| 631 | t.Fatal("a denied HeadObject was reported as a recorded verdict") |
| 632 | } |
| 633 | if done[0].Scanned { |
| 634 | t.Fatal("the engine was asked to read an object with no version") |
| 635 | } |
| 636 | } |
| 637 | |
| 638 | // §7.5: deletion is driven by ABSENCE, never by error. A bucket that fails one |
| 639 | // GET must not cost a tenant an accepted request. |
| 640 | func TestDrainAppKeepsItemOnTransientReadError(t *testing.T) { |
| 641 | ctx := context.Background() |
| 642 | s, sc, idx, _ := queueFixture(t, 1) |
| 643 | enqueueN(t, s, idx, 1) |
| 644 | |
| 645 | flaky := &queueStore{Store: s, failGetPrefix: QueuePrefix("bab", "titogo")} |
| 646 | done, listedEmpty, err := DrainApp(ctx, flaky, sc, idx, "bab", "titogo", 10, time.Now()) |
| 647 | if err != nil { |
| 648 | t.Fatalf("one unreadable item failed the whole drain: %v", err) |
| 649 | } |
| 650 | if len(done) != 0 { |
| 651 | t.Fatalf("outcomes for an item that was never read: %+v", done) |
| 652 | } |
| 653 | if listedEmpty { |
| 654 | t.Fatal("a page with an item in it reported itself empty") |
| 655 | } |
| 656 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 657 | if err != nil || len(left) != 1 { |
| 658 | t.Fatalf("a transient read error discarded the request: %v, %v", left, err) |
| 659 | } |
| 660 | if idx.Len() != 1 { |
| 661 | t.Fatal("the entry pointing at the kept item was released") |
| 662 | } |
| 663 | |
| 664 | // And it drains on the next tick, once the bucket is well again. |
| 665 | done, _, err = DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 666 | if err != nil || len(done) != 1 { |
| 667 | t.Fatalf("retry = %+v, %v", done, err) |
| 668 | } |
| 669 | } |
| 670 | |
| 671 | // The record itself failing to read is reversible too, and it is the SIBLING |
| 672 | // of the check above: QueuePermitted honours "absence, not error" for the |
| 673 | // record, and an earlier draft honoured it there and nowhere else. |
| 674 | func TestDrainAppKeepsItemsOnRecordReadError(t *testing.T) { |
| 675 | ctx := context.Background() |
| 676 | s, sc, idx, _ := queueFixture(t, 1) |
| 677 | enqueueN(t, s, idx, 1) |
| 678 | |
| 679 | broken := &queueStore{Store: s, failGetKey: Key("bab", "titogo")} |
| 680 | if _, _, err := DrainApp(ctx, broken, sc, idx, "bab", "titogo", 10, time.Now()); err == nil { |
| 681 | t.Fatal("an unreadable record reported a successful drain") |
| 682 | } |
| 683 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 684 | if err != nil || len(left) != 1 { |
| 685 | t.Fatalf("an unreadable record discarded the queue: %v, %v", left, err) |
| 686 | } |
| 687 | if idx.Len() != 1 { |
| 688 | t.Fatal("an unreadable record released the index entry") |
| 689 | } |
| 690 | } |
| 691 | |
| 692 | // Engine drift keeps the items too: the record naming a different engine is a |
| 693 | // thing an operator reverses, not a fact about these objects. |
| 694 | func TestDrainAppKeepsItemsOnEngineDrift(t *testing.T) { |
| 695 | ctx := context.Background() |
| 696 | s, sc, idx, _ := queueFixture(t, 1) |
| 697 | enqueueN(t, s, idx, 1) |
| 698 | rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| 699 | rec.Scanning.Enabled.Engine = "someone-elses-scanner" |
| 700 | if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| 701 | t.Fatal(err) |
| 702 | } |
| 703 | |
| 704 | _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 705 | if !errors.Is(err, ErrEngineDrift) { |
| 706 | t.Fatalf("err = %v, want ErrEngineDrift", err) |
| 707 | } |
| 708 | left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 709 | if lerr != nil || len(left) != 1 { |
| 710 | t.Fatalf("drift discarded the queue: %v, %v", left, lerr) |
| 711 | } |
| 712 | if idx.Len() != 1 { |
| 713 | t.Fatal("drift released the index entry") |
| 714 | } |
| 715 | } |
| 716 | |
| 717 | // An engine that will not say what it is keeps them as well — the same |
| 718 | // reasoning as an engine that will not scan. |
| 719 | func TestDrainAppKeepsItemsOnProvenanceFailure(t *testing.T) { |
| 720 | ctx := context.Background() |
| 721 | s, sc, idx, _ := queueFixture(t, 1) |
| 722 | enqueueN(t, s, idx, 1) |
| 723 | sc.Fail("Provenance", errors.New("dial: connection refused")) |
| 724 | |
| 725 | _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 726 | if !errors.Is(err, ErrEngineUnreachable) { |
| 727 | t.Fatalf("err = %v, want ErrEngineUnreachable", err) |
| 728 | } |
| 729 | left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 730 | if lerr != nil || len(left) != 1 { |
| 731 | t.Fatalf("an unreachable engine discarded the queue: %v, %v", left, lerr) |
| 732 | } |
| 733 | } |
| 734 | |
| 735 | // The one read failure that IS permanent: bytes no build will ever parse. |
| 736 | // Leaving it would hold the app's index entry open forever. |
| 737 | func TestDrainAppDeletesUnreadableItem(t *testing.T) { |
| 738 | ctx := context.Background() |
| 739 | s, sc, idx, _ := queueFixture(t, 1) |
| 740 | enqueueN(t, s, idx, 1) |
| 741 | if _, err := s.Put(ctx, QueueKey("bab", "titogo", "0000000000000000000000000v"), |
| 742 | []byte("{not json"), ""); err != nil { |
| 743 | t.Fatal(err) |
| 744 | } |
| 745 | |
| 746 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 747 | if err != nil { |
| 748 | t.Fatal(err) |
| 749 | } |
| 750 | if len(done) != 1 { |
| 751 | t.Fatalf("the readable item did not drain alongside the broken one: %+v", done) |
| 752 | } |
| 753 | left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 754 | if err != nil || len(left) != 0 { |
| 755 | t.Fatalf("the unparseable item was kept: %v, %v", left, err) |
| 756 | } |
| 757 | } |
| 758 | |
| 759 | // A result object is a verdict a door may answer from. When the app may no |
| 760 | // longer be scanned at all, that verdict must go with the queue items — and it |
| 761 | // is the worse orphan of the two: an item is drained next tick, a result is |
| 762 | // found by nothing, blocks the app-name claim gate, and resurrects the app. |
| 763 | func TestDrainAppDeletesResultsWhenNotPermitted(t *testing.T) { |
| 764 | ctx := context.Background() |
| 765 | s, sc, idx, _ := queueFixture(t, 1) |
| 766 | enqueueN(t, s, idx, 1) |
| 767 | if err := WriteScanResult(ctx, s, "bab", "titogo", &ScanResult{ |
| 768 | Key: objectName(0), Verdict: "clean", Version: "v1", At: time.Now().UTC(), |
| 769 | }); err != nil { |
| 770 | t.Fatal(err) |
| 771 | } |
| 772 | rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| 773 | rec.Scanning.Enabled = nil |
| 774 | if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| 775 | t.Fatal(err) |
| 776 | } |
| 777 | |
| 778 | if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); !errors.Is(err, ErrQueueNotPermitted) { |
| 779 | t.Fatalf("drain = %v, want ErrQueueNotPermitted", err) |
| 780 | } |
| 781 | left, err := s.List(ctx, ScanResultsPrefix("bab", "titogo")) |
| 782 | if err != nil || len(left) != 0 { |
| 783 | t.Fatalf("a result outlived the consent that allowed it: %v, %v", left, err) |
| 784 | } |
| 785 | } |
| 786 | |
| 787 | // A store that was never granted has no bucket to scan. The check has to be |
| 788 | // the FIRST one, because everything after it dereferences Provisioned. |
| 789 | func TestDrainAppRefusesWhenNotGranted(t *testing.T) { |
| 790 | ctx := context.Background() |
| 791 | |
| 792 | for name, break_ := range map[string]func(t *testing.T, s release.Store){ |
| 793 | "record absent": func(t *testing.T, s release.Store) { |
| 794 | if err := s.Delete(ctx, Key("bab", "titogo")); err != nil { |
| 795 | t.Fatal(err) |
| 796 | } |
| 797 | }, |
| 798 | "declared but never granted": func(t *testing.T, s release.Store) { |
| 799 | rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| 800 | rec.Provisioned = nil |
| 801 | if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| 802 | t.Fatal(err) |
| 803 | } |
| 804 | }, |
| 805 | } { |
| 806 | t.Run(name, func(t *testing.T) { |
| 807 | s, sc, idx, _ := queueFixture(t, 1) |
| 808 | enqueueN(t, s, idx, 1) |
| 809 | break_(t, s) |
| 810 | |
| 811 | _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 812 | if !errors.Is(err, ErrQueueNotPermitted) { |
| 813 | t.Fatalf("err = %v, want ErrQueueNotPermitted", err) |
| 814 | } |
| 815 | left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 816 | if lerr != nil || len(left) != 0 { |
| 817 | t.Fatalf("items survived a permanent refusal: %v, %v", left, lerr) |
| 818 | } |
| 819 | }) |
| 820 | } |
| 821 | } |
| 822 | |
| 823 | // A drain with no engine refuses in words rather than panicking on a nil |
| 824 | // interface, the way RunSweep does. The goroutine gates on this too, but a |
| 825 | // package that dereferences whatever it is handed is one call site away from |
| 826 | // taking the console down. |
| 827 | func TestDrainAppRefusesWithoutAnEngine(t *testing.T) { |
| 828 | ctx := context.Background() |
| 829 | s, _, idx, _ := queueFixture(t, 1) |
| 830 | enqueueN(t, s, idx, 1) |
| 831 | |
| 832 | _, _, err := DrainApp(ctx, s, nil, idx, "bab", "titogo", 10, time.Now()) |
| 833 | if !errors.Is(err, ErrNoEngine) { |
| 834 | t.Fatalf("err = %v, want ErrNoEngine", err) |
| 835 | } |
| 836 | left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 837 | if lerr != nil || len(left) != 1 { |
| 838 | t.Fatalf("a missing engine discarded the queue: %v, %v", left, lerr) |
| 839 | } |
| 840 | } |
| 841 | |
| 842 | // The purge race, at the drain. The re-read before writing a result is the |
| 843 | // only thing between an unrecordable verdict and an orphan under a purged |
| 844 | // prefix — and that orphan does not merely leak: the claim gate refuses the |
| 845 | // app's name for good, and materialize counts any key under the prefix and |
| 846 | // fails toward keeping the row, so the deleted app comes BACK on the next boot. |
| 847 | func TestResultObjectNotWrittenUnderPurgedPrefix(t *testing.T) { |
| 848 | ctx := context.Background() |
| 849 | s, sc, idx, _ := queueFixture(t, 1) |
| 850 | sc.Unrecordable(objectName(0)) // the ten-tag cap, in production |
| 851 | enqueueN(t, s, idx, 1) |
| 852 | |
| 853 | // The teardown lands between the PER-OBJECT consent check (the second read |
| 854 | // of store.json, added when a review pointed out that one check per page |
| 855 | // is not one check per object) and drainOne's re-read before writing (the |
| 856 | // third). The re-read is the guard being tested here; without it the |
| 857 | // create-only write SUCCEEDS under a prefix nothing will ever list again. |
| 858 | // |
| 859 | // The count is 3 rather than 2 because there are now three reads of this |
| 860 | // key in a drain, not two. If it ever drifts again, the tell is this test |
| 861 | // failing with ErrQueueNotPermitted — that means the race moved to an |
| 862 | // EARLIER guard, which is the case TestPurgeBeforeTheObjectIsReadIsRefused |
| 863 | // below covers on purpose. |
| 864 | staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 3, race: func() { |
| 865 | // What PurgeApp leaves behind: the prefix, emptied. Done directly |
| 866 | // because PurgeApp itself refuses over a provisioned record, and the |
| 867 | // teardown that clears it is not what this test is about. |
| 868 | keys, err := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| 869 | if err != nil { |
| 870 | t.Error(err) |
| 871 | return |
| 872 | } |
| 873 | for _, k := range keys { |
| 874 | if err := s.Delete(ctx, k); err != nil { |
| 875 | t.Error(err) |
| 876 | } |
| 877 | } |
| 878 | }} |
| 879 | |
| 880 | done, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now()) |
| 881 | // REPORTED, not swallowed. The first version returned (out, nil) here, so |
| 882 | // the drain read the group as a success: it deleted the items, put an |
| 883 | // outcome in done, and carried on scanning the rest of the page knowing |
| 884 | // consent was gone. The outcome it produced had Recorded false with no |
| 885 | // result object behind it, which task 16 would have turned into a callback |
| 886 | // saying "clean" about an object the status door answers "pending" about |
| 887 | // forever. |
| 888 | if !errors.Is(err, ErrQueueNotPermitted) { |
| 889 | t.Fatalf("drain err = %v, want ErrQueueNotPermitted", err) |
| 890 | } |
| 891 | if len(done) != 0 { |
| 892 | t.Errorf("done carries %d outcome(s) for a store that may not be scanned: %+v", len(done), done) |
| 893 | } |
| 894 | left, lerr := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| 895 | if lerr != nil { |
| 896 | t.Fatal(lerr) |
| 897 | } |
| 898 | if len(left) != 0 { |
| 899 | t.Fatalf("the purged app's prefix still holds %v — its name can never be claimed again", left) |
| 900 | } |
| 901 | } |
| 902 | |
| 903 | // TestPurgeBeforeTheObjectIsReadIsRefused is the guard the per-object consent |
| 904 | // check adds, and it is the better of the two outcomes: the teardown lands |
| 905 | // before this object's bytes are read at all, so nothing is fetched, nothing |
| 906 | // is scanned and no result can be stranded. |
| 907 | // |
| 908 | // The page-level check alone could not do this. It runs once and then the loop |
| 909 | // reads up to fifty objects, so a purge — or a revoked grant, or a member |
| 910 | // re-attesting the store as client-encrypted — was honoured only from the next |
| 911 | // tick, and every object still in the page was read under a consent that had |
| 912 | // already been withdrawn. Found by an external security review. |
| 913 | func TestPurgeBeforeTheObjectIsReadIsRefused(t *testing.T) { |
| 914 | ctx := context.Background() |
| 915 | s, sc, idx, _ := queueFixture(t, 1) |
| 916 | enqueueN(t, s, idx, 1) |
| 917 | |
| 918 | staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 2, race: func() { |
| 919 | keys, err := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| 920 | if err != nil { |
| 921 | t.Error(err) |
| 922 | return |
| 923 | } |
| 924 | for _, k := range keys { |
| 925 | if err := s.Delete(ctx, k); err != nil { |
| 926 | t.Error(err) |
| 927 | } |
| 928 | } |
| 929 | }} |
| 930 | |
| 931 | _, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now()) |
| 932 | if !errors.Is(err, ErrQueueNotPermitted) { |
| 933 | t.Fatalf("err = %v, want ErrQueueNotPermitted — consent was gone before this object was read", err) |
| 934 | } |
| 935 | if n := sc.Scans(objectName(0)); n != 0 { |
| 936 | t.Errorf("scanned %d time(s) after consent was withdrawn mid-page", n) |
| 937 | } |
| 938 | left, lerr := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| 939 | if lerr != nil { |
| 940 | t.Fatal(lerr) |
| 941 | } |
| 942 | if len(left) != 0 { |
| 943 | t.Fatalf("the purged app's prefix still holds %v", left) |
| 944 | } |
| 945 | } |
| 946 | |
| 947 | // recordRaceStore runs race() just before the Nth read of one key, so a test |
| 948 | // can put a competing writer at a chosen point in a walk. Same shape as |
| 949 | // walk_test.go's raceOnReadStore, which fires on the first read only. |
| 950 | type recordRaceStore struct { |
| 951 | release.Store |
| 952 | key string |
| 953 | before int // fire immediately before this read, 1-based |
| 954 | race func() |
| 955 | seen int |
| 956 | } |
| 957 | |
| 958 | func (r *recordRaceStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 959 | if key == r.key { |
| 960 | r.seen++ |
| 961 | if r.seen == r.before { |
| 962 | r.race() |
| 963 | } |
| 964 | } |
| 965 | return r.Store.Get(ctx, key) |
| 966 | } |
| 967 | |
| 968 | func (r *recordRaceStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| 969 | return r.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit) |
| 970 | } |
| 971 | |
| 972 | // THE CONSENT CHAIN IS ONE FUNCTION AND THIS IS THE TABLE THAT SAYS SO. |
| 973 | // |
| 974 | // Spec §5.1.1: the enqueue door and the drain must both refuse unless every |
| 975 | // link holds, and the reason they must not drift is that they have different |
| 976 | // consequences for the same answer — the door turns it into an HTTP state, the |
| 977 | // drain turns it into keep-or-delete (§7.5). A bool cannot carry that. It has |
| 978 | // already been a bool through ten tasks, and every caller reconstructed the |
| 979 | // reason from the record it was handed back, which is exactly the shape where |
| 980 | // two callers start disagreeing. |
| 981 | // |
| 982 | // So the refusal is named, and the two consequences are read off the name. The |
| 983 | // split this asserts is §7.5's, verbatim: everything upstream of the |
| 984 | // deployment switch is permanent, because none of it can change back in a way |
| 985 | // that makes THIS item useful; the switch itself is reversible, because an |
| 986 | // operator turns it on again and the item is still worth draining. |
| 987 | func TestQueueRefusalNamesEachLinkAndItsConsequence(t *testing.T) { |
| 988 | ctx := context.Background() |
| 989 | |
| 990 | cases := []struct { |
| 991 | name string |
| 992 | // break mutates a fully permitted fixture to snap one link. |
| 993 | snap func(t *testing.T, s release.Store) |
| 994 | want QueueRefusal |
| 995 | permanent bool |
| 996 | }{ |
| 997 | { |
| 998 | name: "nothing is broken", |
| 999 | snap: func(*testing.T, release.Store) {}, |
| 1000 | want: RefusalNone, |
| 1001 | }, |
| 1002 | { |
| 1003 | name: "the store was never granted", |
| 1004 | snap: func(t *testing.T, s release.Store) { |
| 1005 | mutateRecord(t, s, func(r *Record) { r.Provisioned = nil }) |
| 1006 | }, |
| 1007 | want: RefusalNotGranted, permanent: true, |
| 1008 | }, |
| 1009 | { |
| 1010 | name: "the app is in the trash", |
| 1011 | snap: func(t *testing.T, s release.Store) { |
| 1012 | if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil { |
| 1013 | t.Fatal(err) |
| 1014 | } |
| 1015 | }, |
| 1016 | want: RefusalTrashed, permanent: true, |
| 1017 | }, |
| 1018 | { |
| 1019 | name: "the member never requested scanning", |
| 1020 | snap: func(t *testing.T, s release.Store) { |
| 1021 | mutateRecord(t, s, func(r *Record) { r.Scanning.Requested = nil }) |
| 1022 | }, |
| 1023 | want: RefusalNotRequested, permanent: true, |
| 1024 | }, |
| 1025 | { |
| 1026 | name: "the member attested the store client-encrypted", |
| 1027 | snap: func(t *testing.T, s release.Store) { |
| 1028 | mutateRecord(t, s, func(r *Record) { r.Scanning.Requested.Encrypted = true }) |
| 1029 | }, |
| 1030 | want: RefusalEncrypted, permanent: true, |
| 1031 | }, |
| 1032 | { |
| 1033 | name: "the operator never granted scanning", |
| 1034 | snap: func(t *testing.T, s release.Store) { |
| 1035 | mutateRecord(t, s, func(r *Record) { r.Scanning.Enabled = nil }) |
| 1036 | }, |
| 1037 | want: RefusalNotEnabled, permanent: true, |
| 1038 | }, |
| 1039 | { |
| 1040 | // The one reversible link, and the reason Permanent() exists at |
| 1041 | // all: an operator runs `carlos system antivirus enable` and every |
| 1042 | // item still in the queue is worth draining. Delete these and a |
| 1043 | // deployment that was off for ten minutes loses a tenant's |
| 1044 | // accepted requests. |
| 1045 | name: "the deployment switch is off", |
| 1046 | snap: func(t *testing.T, s release.Store) { |
| 1047 | disableDeployment(t, s) |
| 1048 | }, |
| 1049 | want: RefusalDisabled, permanent: false, |
| 1050 | }, |
| 1051 | } |
| 1052 | |
| 1053 | for _, tc := range cases { |
| 1054 | t.Run(tc.name, func(t *testing.T) { |
| 1055 | s, _, _, _ := queueFixture(t, 1) |
| 1056 | tc.snap(t, s) |
| 1057 | |
| 1058 | rec, refusal, err := QueuePermitted(ctx, s, "bab", "titogo") |
| 1059 | if err != nil { |
| 1060 | t.Fatalf("QueuePermitted: %v", err) |
| 1061 | } |
| 1062 | if refusal != tc.want { |
| 1063 | t.Fatalf("refusal = %q, want %q", refusal, tc.want) |
| 1064 | } |
| 1065 | if refusal.Permanent() != tc.permanent { |
| 1066 | t.Fatalf("%q.Permanent() = %v, want %v — this decides whether the drain deletes the item or keeps it (§7.5)", |
| 1067 | refusal, refusal.Permanent(), tc.permanent) |
| 1068 | } |
| 1069 | if tc.want == RefusalNone && rec == nil { |
| 1070 | t.Fatal("a permitted app came back with no record") |
| 1071 | } |
| 1072 | }) |
| 1073 | } |
| 1074 | } |
| 1075 | |
| 1076 | // RefusalNone must never be Permanent: "keep going" is not a deletion reason, |
| 1077 | // and a Permanent() written as `r != RefusalDisabled` would say it was. |
| 1078 | func TestPermittedIsNotAPermanentRefusal(t *testing.T) { |
| 1079 | if RefusalNone.Permanent() { |
| 1080 | t.Fatal("RefusalNone.Permanent() is true — the drain would delete every item it was allowed to scan") |
| 1081 | } |
| 1082 | } |
| 1083 | |
| 1084 | // A READ ERROR IS NOT A REFUSAL. It has no name in the chain because nothing |
| 1085 | // about consent was learned: §7.5's "deletion is driven by absence, never by |
| 1086 | // error". The error comes back as an error, and the refusal stays empty — so a |
| 1087 | // caller that forgets to check err cannot mistake a bad bucket moment for |
| 1088 | // permission, and equally cannot mistake it for a permanent no. |
| 1089 | func TestQueuePermittedReportsAReadErrorAsAnError(t *testing.T) { |
| 1090 | s, _, _, _ := queueFixture(t, 1) |
| 1091 | broken := &queueStore{Store: s, failGetKey: "apps/bab/titogo/store.json"} |
| 1092 | |
| 1093 | rec, refusal, err := QueuePermitted(context.Background(), broken, "bab", "titogo") |
| 1094 | if err == nil { |
| 1095 | t.Fatal("a failing store.json read came back as a clean answer") |
| 1096 | } |
| 1097 | if refusal != RefusalNone { |
| 1098 | t.Fatalf("refusal = %q on a read error, want the empty refusal — a named refusal here would be deleted on (§7.5)", refusal) |
| 1099 | } |
| 1100 | if rec != nil { |
| 1101 | t.Fatal("a record came back from a read that failed") |
| 1102 | } |
| 1103 | } |
| 1104 | |
| 1105 | // mutateRecord reads, edits and writes back one store.json with the real |
| 1106 | // etag, so a test can snap exactly one link of the consent chain and leave |
| 1107 | // every other link intact. |
| 1108 | func mutateRecord(t *testing.T, s release.Store, edit func(*Record)) { |
| 1109 | t.Helper() |
| 1110 | ctx := context.Background() |
| 1111 | rec, etag, err := Get(ctx, s, "bab", "titogo") |
| 1112 | if err != nil || rec == nil { |
| 1113 | t.Fatalf("re-read the record: %+v, %v", rec, err) |
| 1114 | } |
| 1115 | edit(rec) |
| 1116 | if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| 1117 | t.Fatal(err) |
| 1118 | } |
| 1119 | } |
| 1120 | |
| 1121 | // disableDeployment turns the deployment switch off the way `carlos system |
| 1122 | // antivirus disable` does — the record, not an env var. |
| 1123 | func disableDeployment(t *testing.T, s release.Store) { |
| 1124 | t.Helper() |
| 1125 | ctx := context.Background() |
| 1126 | av, etag, err := release.GetAntivirus(ctx, s) |
| 1127 | if err != nil || av == nil { |
| 1128 | t.Fatalf("antivirus record: %+v, %v", av, err) |
| 1129 | } |
| 1130 | av.Enabled = false |
| 1131 | if _, err := release.PutAntivirus(ctx, s, av, etag); err != nil { |
| 1132 | t.Fatal(err) |
| 1133 | } |
| 1134 | } |
| 1135 | |
| 1136 | // readErrorStore fails Get for one exact key, and behaves for everything else. |
| 1137 | // A transient bad moment on one object, which is the case §7.5 separates from |
| 1138 | // absence. |
| 1139 | type readErrorStore struct { |
| 1140 | release.Store |
| 1141 | key string |
| 1142 | err error |
| 1143 | } |
| 1144 | |
| 1145 | func (r *readErrorStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 1146 | if key == r.key { |
| 1147 | return nil, "", r.err |
| 1148 | } |
| 1149 | return r.Store.Get(ctx, key) |
| 1150 | } |
| 1151 | |
| 1152 | // TestRebuildKeepsAnItemItCouldNotRead pins the claim RebuildQueue's own |
| 1153 | // comment makes and nothing tested: a STORE error is the store having a bad |
| 1154 | // moment, so the item counts as live and the index still points at it. |
| 1155 | // |
| 1156 | // The opposite — treating an unreadable item as gone — silently strands it, |
| 1157 | // because this rebuild is the only thing that would ever have listed that |
| 1158 | // prefix again. §7.5's rule is that deletion is driven by absence, never by |
| 1159 | // error, and this is that rule on the read side. |
| 1160 | func TestRebuildKeepsAnItemItCouldNotRead(t *testing.T) { |
| 1161 | ctx := context.Background() |
| 1162 | s, _, idx, _ := queueFixture(t, 1) |
| 1163 | enqueueN(t, s, idx, 1) |
| 1164 | |
| 1165 | keys, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 1166 | if err != nil || len(keys) != 1 { |
| 1167 | t.Fatalf("staging: %v %v", keys, err) |
| 1168 | } |
| 1169 | broken := &readErrorStore{Store: s, key: keys[0], err: errors.New("s3 had a moment")} |
| 1170 | |
| 1171 | fresh := NewQueueIndex() |
| 1172 | live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now()) |
| 1173 | if err != nil { |
| 1174 | t.Fatalf("a read error failed the whole rebuild: %v", err) |
| 1175 | } |
| 1176 | if live != 1 { |
| 1177 | t.Errorf("live = %d, want 1 — an item we could not read is not an item we know is gone", live) |
| 1178 | } |
| 1179 | if fresh.Len() != 1 { |
| 1180 | t.Error("the index does not point at the app, so nothing will ever list that prefix again") |
| 1181 | } |
| 1182 | left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| 1183 | if lerr != nil || len(left) != 1 { |
| 1184 | t.Errorf("the item was deleted on a store error: %v %v", left, lerr) |
| 1185 | } |
| 1186 | } |
| 1187 | |
| 1188 | // TestRebuildStillIndexesWhenADeleteFails is the blocker an adversarial review |
| 1189 | // found: the bump used to happen only on the success path, so one failed |
| 1190 | // Delete of an EXPIRED item returned early and left the app's LIVE items with |
| 1191 | // nothing pointing at them — until the next restart, since this pass is the |
| 1192 | // only thing that collects a stranded item. |
| 1193 | func TestRebuildStillIndexesWhenADeleteFails(t *testing.T) { |
| 1194 | ctx := context.Background() |
| 1195 | s, _, idx, _ := queueFixture(t, 2) |
| 1196 | |
| 1197 | // One expired item and one live one. |
| 1198 | if _, err := Enqueue(ctx, s, idx, "bab", "titogo", |
| 1199 | &QueueItem{Key: objectName(0), By: "app"}, time.Now().Add(-QueueTTL-time.Hour)); err != nil { |
| 1200 | t.Fatal(err) |
| 1201 | } |
| 1202 | if _, err := Enqueue(ctx, s, idx, "bab", "titogo", |
| 1203 | &QueueItem{Key: objectName(1), By: "app"}, time.Now()); err != nil { |
| 1204 | t.Fatal(err) |
| 1205 | } |
| 1206 | broken := &deleteErrorStore{Store: s, err: errors.New("s3 refused the delete")} |
| 1207 | |
| 1208 | fresh := NewQueueIndex() |
| 1209 | live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now()) |
| 1210 | if err == nil { |
| 1211 | t.Fatal("the failed delete was not reported") |
| 1212 | } |
| 1213 | _ = live |
| 1214 | if fresh.Len() != 1 { |
| 1215 | t.Fatalf("index holds %d app(s) after a failed delete, want 1 — the live item is stranded until the next restart", fresh.Len()) |
| 1216 | } |
| 1217 | } |
| 1218 | |
| 1219 | // deleteErrorStore fails every Delete. |
| 1220 | type deleteErrorStore struct { |
| 1221 | release.Store |
| 1222 | err error |
| 1223 | } |
| 1224 | |
| 1225 | func (d *deleteErrorStore) Delete(context.Context, string) error { return d.err } |
| 1226 | |
| 1227 | // TestReusedVerdictCarriesNoProvenance is the one an adversarial review found |
| 1228 | // missing, and the defect it pins was live: the outcome was stamped with the |
| 1229 | // engine and signature database THIS TICK is configured with, before anything |
| 1230 | // knew whether the engine had run. |
| 1231 | // |
| 1232 | // The scenario is ordinary. The hourly sweep tags an object clean under |
| 1233 | // signature database 27100. A week later the app asks for a scan; the drain |
| 1234 | // reads the tag, does not scan, and — before this — reported engine "clamav" |
| 1235 | // and today's database 27412. That outcome becomes a SIGNED callback body, |
| 1236 | // so the platform would have put its name to a claim about which database |
| 1237 | // judged an object when no such judgement happened. The status door already |
| 1238 | // refuses to make that guess, so the two surfaces disagreed and the signed |
| 1239 | // one was the one lying. |
| 1240 | // |
| 1241 | // Empty is the honest answer. A verdict with no provenance is useful; a |
| 1242 | // verdict with invented provenance is the failure the feature exists to stop. |
| 1243 | func TestReusedVerdictCarriesNoProvenance(t *testing.T) { |
| 1244 | ctx := context.Background() |
| 1245 | s, sc, idx, bucket := queueFixture(t, 1) |
| 1246 | |
| 1247 | // Pre-tag the object at its current version, as a sweep would have. |
| 1248 | obj := ScanObject{Key: objectName(0), ETag: "etag-" + objectName(0), Size: 10} |
| 1249 | status, seen, err := sc.Status(ctx, bucket, objectName(0)) |
| 1250 | if err != nil { |
| 1251 | t.Fatal(err) |
| 1252 | } |
| 1253 | _ = status |
| 1254 | obj = seen |
| 1255 | sc.Put(bucket, objectName(0), obj.Version(), VerdictClean) |
| 1256 | enqueueN(t, s, idx, 1) |
| 1257 | |
| 1258 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 1259 | if err != nil { |
| 1260 | t.Fatal(err) |
| 1261 | } |
| 1262 | if len(done) != 1 { |
| 1263 | t.Fatalf("outcomes = %+v, want one", done) |
| 1264 | } |
| 1265 | got := done[0] |
| 1266 | if got.Verdict != VerdictClean || !got.Recorded { |
| 1267 | t.Fatalf("outcome = %+v, want the tag's clean verdict", got) |
| 1268 | } |
| 1269 | if got.Scanned { |
| 1270 | t.Fatal("the engine ran for an object that already carried a verdict") |
| 1271 | } |
| 1272 | if got.Engine != "" || got.Signature != "" { |
| 1273 | t.Errorf("reused verdict claims engine %q signature %q — neither judged this object", |
| 1274 | got.Engine, got.Signature) |
| 1275 | } |
| 1276 | } |
| 1277 | |
| 1278 | // TestScannedVerdictCarriesProvenance is the other half, without which the |
| 1279 | // test above passes against a drain that never reports provenance at all. |
| 1280 | func TestScannedVerdictCarriesProvenance(t *testing.T) { |
| 1281 | ctx := context.Background() |
| 1282 | s, sc, idx, _ := queueFixture(t, 1) |
| 1283 | enqueueN(t, s, idx, 1) |
| 1284 | |
| 1285 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 1286 | if err != nil { |
| 1287 | t.Fatal(err) |
| 1288 | } |
| 1289 | if len(done) != 1 || !done[0].Scanned { |
| 1290 | t.Fatalf("outcomes = %+v, want one that was actually scanned", done) |
| 1291 | } |
| 1292 | if done[0].Engine == "" { |
| 1293 | t.Error("an outcome the engine really produced carries no engine name") |
| 1294 | } |
| 1295 | } |
| 1296 | |
| 1297 | // TestDrainMetersOnlyBytesTheEngineRead pins Scanned to the bytes reaching |
| 1298 | // the engine, not to the Scan call having been made. A skipped object is one |
| 1299 | // over the size ceiling that was never read, and a denied one is a GET the |
| 1300 | // console was refused; both used to count their full size against the tenant. |
| 1301 | // The meter is unpriced and best-effort, but it must not overstate. |
| 1302 | func TestDrainMetersOnlyBytesTheEngineRead(t *testing.T) { |
| 1303 | ctx := context.Background() |
| 1304 | s, sc, idx, _ := queueFixture(t, 4) |
| 1305 | sc.ScanAs(objectName(1), VerdictSkipped) |
| 1306 | sc.ScanAs(objectName(2), VerdictDenied) |
| 1307 | sc.ScanAs(objectName(3), VerdictInfected) |
| 1308 | enqueueN(t, s, idx, 4) |
| 1309 | |
| 1310 | done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| 1311 | if err != nil { |
| 1312 | t.Fatal(err) |
| 1313 | } |
| 1314 | byKey := map[string]Drained{} |
| 1315 | for _, d := range done { |
| 1316 | byKey[d.Key] = d |
| 1317 | } |
| 1318 | if len(byKey) != 4 { |
| 1319 | t.Fatalf("outcomes = %+v, want four", done) |
| 1320 | } |
| 1321 | for _, key := range []string{objectName(1), objectName(2)} { |
| 1322 | d := byKey[key] |
| 1323 | if n := sc.Scans(key); n != 1 { |
| 1324 | t.Fatalf("%s: the engine was asked %d time(s), want once — this test is about an answer the call gave", key, n) |
| 1325 | } |
| 1326 | if d.Scanned { |
| 1327 | t.Errorf("%s answered %q and is metered as scanned: %d bytes nothing read would be charged", key, d.Verdict, d.Size) |
| 1328 | } |
| 1329 | } |
| 1330 | for _, key := range []string{objectName(0), objectName(3)} { |
| 1331 | d := byKey[key] |
| 1332 | if !d.Scanned || d.Size != 10 { |
| 1333 | t.Errorf("%s answered %q with Scanned=%v Size=%d, want a scanned object of 10 bytes", key, d.Verdict, d.Scanned, d.Size) |
| 1334 | } |
| 1335 | } |
| 1336 | } |
| 1337 | |
| 1338 | // TestRebuildWithNothingStaleDeletesNothing: the rebuild reads every item and |
| 1339 | // enters its collect only when it found something to collect. Over a store |
| 1340 | // whose every Delete fails, a prefix of live items rebuilds without error, |
| 1341 | // which is only possible if no delete was attempted. |
| 1342 | func TestRebuildWithNothingStaleDeletesNothing(t *testing.T) { |
| 1343 | ctx := context.Background() |
| 1344 | s, _, idx, _ := queueFixture(t, 2) |
| 1345 | enqueueN(t, s, idx, 2) |
| 1346 | broken := &deleteErrorStore{Store: s, err: errors.New("no delete may happen here")} |
| 1347 | |
| 1348 | fresh := NewQueueIndex() |
| 1349 | live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now()) |
| 1350 | if err != nil { |
| 1351 | t.Fatalf("a rebuild with nothing to collect reached a delete: %v", err) |
| 1352 | } |
| 1353 | if live != 2 || fresh.Len() != 1 { |
| 1354 | t.Fatalf("live = %d, indexed apps = %d, want 2 and 1", live, fresh.Len()) |
| 1355 | } |
| 1356 | } |
| 1357 | |
amadan