amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "os"
9 "path/filepath"
10 "strings"
11 "sync/atomic"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/objstore"
16 "github.com/carlosframework/platform/internal/release"
17 "github.com/carlosframework/platform/internal/usage"
18)
19
20// drainFixture is a console whose release store is a PLAIN DirStore.
21//
22// The shared fixture wraps its store in a struct that embeds release.Store as
23// an INTERFACE, which hides the two optional interfaces this whole file is
24// about: PageLister, which is how the drain caps one app's turn (§7.3), and
25// ModLister, which is how expired result objects are collected. A test on the
26// wrapper would silently exercise the degraded fallback in both cases and
27// prove nothing about the path a deployment actually runs.
28//
29// It returns the bucket root because two tests have to AGE a file, and mtime
30// is what DirStore's ListMod reads.
31func drainFixture(t *testing.T) (*consoleFixture, *objstore.MemScanner, string) {
32 t.Helper()
33 fx := newConsoleFixture(t)
34 dir := t.TempDir()
35 rel := release.NewDirStore(dir)
36 fx.rel = rel
37 fx.srv.cfg.Releases = rel
38 sc := objstore.NewMemScanner()
39 fx.srv.cfg.StoreScanner = sc
40 return fx, sc, dir
41}
42
43func drainBucket(acct, app string) string { return "carlos-" + acct + "-" + app }
44
45// enqueueFor puts one item in the bucket AND in the server's index, which is
46// what the door does.
47func enqueueFor(t *testing.T, fx *consoleFixture, acct, app, key string, at time.Time) {
48 t.Helper()
49 if _, err := objstore.Enqueue(context.Background(), fx.rel, fx.srv.scanQueue, acct, app,
50 &objstore.QueueItem{Key: key, By: "test"}, at); err != nil {
51 t.Fatalf("enqueue %s: %v", key, err)
52 }
53}
54
55// enqueueStranded is the crash §7.1 describes: the item object is written and
56// the process dies before the index is bumped. Modelled by bumping a THROWAWAY
57// index, so the bucket has the item and this console has never heard of it.
58func enqueueStranded(t *testing.T, fx *consoleFixture, acct, app, key string, at time.Time) {
59 t.Helper()
60 if _, err := objstore.Enqueue(context.Background(), fx.rel, objstore.NewQueueIndex(), acct, app,
61 &objstore.QueueItem{Key: key, By: "test"}, at); err != nil {
62 t.Fatalf("enqueue %s: %v", key, err)
63 }
64}
65
66func queueLen(t *testing.T, fx *consoleFixture, acct, app string) int {
67 t.Helper()
68 n, err := objstore.QueueDepth(context.Background(), fx.rel, acct, app)
69 if err != nil {
70 t.Fatalf("queue depth: %v", err)
71 }
72 return n
73}
74
75func resultKeys(t *testing.T, fx *consoleFixture, acct, app string) []string {
76 t.Helper()
77 keys, err := fx.rel.List(context.Background(), objstore.ScanResultsPrefix(acct, app))
78 if err != nil {
79 t.Fatalf("list results: %v", err)
80 }
81 return keys
82}
83
84// countingStore counts listings, which is the only way to see the difference
85// between "the drain refused" and "the drain did the work and objstore
86// refused".
87//
88// It embeds the CONCRETE *release.DirStore rather than the interface, so
89// ListMod and ListDirs are still promoted — embedding the interface is exactly
90// the trap drainFixture's comment describes.
91type countingStore struct {
92 *release.DirStore
93 lists atomic.Int64
94}
95
96type scanUsageCountingStore struct {
97 *release.DirStore
98 puts atomic.Int64
99}
100
101func (s *scanUsageCountingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
102 if strings.HasPrefix(key, usage.ScanPrefix) {
103 s.puts.Add(1)
104 }
105 return s.DirStore.Put(ctx, key, body, ifMatch)
106}
107
108func (c *countingStore) List(ctx context.Context, prefix string) ([]string, error) {
109 c.lists.Add(1)
110 return c.DirStore.List(ctx, prefix)
111}
112
113func (c *countingStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
114 c.lists.Add(1)
115 return c.DirStore.ListPage(ctx, prefix, startAfter, limit)
116}
117
118// setDeploymentScanning flips the deployment switch either way.
119//
120// enableDeploymentScanning is CREATE-ONLY — it returns early when a record
121// already exists, which is right for a fixture arming a fresh deployment and
122// useless for re-enabling one that was turned off.
123func setDeploymentScanning(t *testing.T, fx *consoleFixture, on bool) {
124 t.Helper()
125 ctx := context.Background()
126 av, etag, err := release.GetAntivirus(ctx, fx.rel)
127 if err != nil {
128 t.Fatal(err)
129 }
130 if av == nil {
131 av = &release.AntivirusSettings{Engine: "clamav"}
132 }
133 av.Enabled = on
134 av.At = time.Now().UTC()
135 av.By = "test"
136 if _, err := release.PutAntivirus(ctx, fx.rel, av, etag); err != nil {
137 t.Fatal(err)
138 }
139}
140
141// revokeScanning is the operator taking the per-store grant away AFTER the
142// record exists, which enableStore cannot do: it writes with an empty etag and
143// so conflicts with anything already there.
144func revokeScanning(t *testing.T, fx *consoleFixture, acct, app string) {
145 t.Helper()
146 ctx := context.Background()
147 rec, etag, err := objstore.Get(ctx, fx.rel, acct, app)
148 if err != nil {
149 t.Fatal(err)
150 }
151 if rec == nil {
152 t.Fatalf("no record for %s/%s to revoke", acct, app)
153 }
154 rec.Scanning = nil
155 if _, err := objstore.Put(ctx, fx.rel, acct, app, rec, etag); err != nil {
156 t.Fatal(err)
157 }
158}
159
160// TestBootRebuildFindsStrandedItem is the whole argument for the rebuild
161// existing, and it is written so that the rebuild is the ONLY difference
162// between the two halves.
163//
164// A tick runs first and must do nothing: the drain only visits apps the index
165// names, and nothing named this one. Then the rebuild runs, and the same tick
166// scans. Without the "before" half this would pass against a drain that walked
167// every app on the deployment every 30s — which is the design the index exists
168// to avoid.
169func TestBootRebuildFindsStrandedItem(t *testing.T) {
170 ctx := context.Background()
171 fx, sc, _ := drainFixture(t)
172 enableStore(t, fx, fixtureAcct, fixtureApp, true)
173 sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
174 enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
175
176 if d := fx.srv.ScanQueueDepth(); d != 0 {
177 t.Fatalf("index depth %d before the rebuild — the item was not stranded, so this test proves nothing", d)
178 }
179 if err := fx.srv.scanDrainTick(ctx); err != nil {
180 t.Fatal(err)
181 }
182 if n := sc.Scans("uploads/x.pdf"); n != 0 {
183 t.Fatalf("a tick scanned %d time(s) for an app the index never named", n)
184 }
185
186 if err := fx.srv.rebuildScanQueue(ctx); err != nil {
187 t.Fatal(err)
188 }
189 if d := fx.srv.ScanQueueDepth(); d != 1 {
190 t.Fatalf("index depth %d after the rebuild, want 1 — nothing would ever list that prefix again", d)
191 }
192 if err := fx.srv.scanDrainTick(ctx); err != nil {
193 t.Fatal(err)
194 }
195 if n := sc.Scans("uploads/x.pdf"); n != 1 {
196 t.Fatalf("scanned %d times after the rebuild, want 1", n)
197 }
198 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
199 t.Errorf("%d item(s) left after a successful drain", n)
200 }
201}
202
203// TestBootRebuildCollectsExpiredItemsAndResults covers BOTH prefixes on the
204// same 24h TTL, because they expire for the same reason: after a sweep cycle
205// the ordinary walk has had its chance at the object, and a stale on-demand
206// answer is worth less than none.
207//
208// The result object matters more than the item. A stranded item is drained on
209// the next tick; a stranded result is found by nothing, blocks the app-name
210// claim gate, and would sit in the bucket forever.
211func TestBootRebuildCollectsExpiredItemsAndResults(t *testing.T) {
212 ctx := context.Background()
213 fx, _, dir := drainFixture(t)
214 enableStore(t, fx, fixtureAcct, fixtureApp, true)
215
216 old := time.Now().Add(-objstore.QueueTTL - time.Hour)
217 enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/stale.pdf", old)
218 if err := objstore.WriteScanResult(ctx, fx.rel, fixtureAcct, fixtureApp, &objstore.ScanResult{
219 Key: "uploads/untaggable.pdf", Verdict: "clean", Version: "etag-9@x",
220 Engine: "clamav", Signature: "sig", At: old.UTC(),
221 }); err != nil {
222 t.Fatal(err)
223 }
224 // ListMod reads MTIME, not the At field inside the JSON, so the file has
225 // to actually be old.
226 resKey := objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/untaggable.pdf")
227 if err := os.Chtimes(filepath.Join(dir, filepath.FromSlash(resKey)), old, old); err != nil {
228 t.Fatal(err)
229 }
230
231 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
232 t.Fatalf("%d item(s) staged, want 1", n)
233 }
234 if got := resultKeys(t, fx, fixtureAcct, fixtureApp); len(got) != 1 {
235 t.Fatalf("%d result(s) staged, want 1", len(got))
236 }
237
238 if err := fx.srv.rebuildScanQueue(ctx); err != nil {
239 t.Fatal(err)
240 }
241 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
242 t.Errorf("%d expired item(s) survived the rebuild", n)
243 }
244 if got := resultKeys(t, fx, fixtureAcct, fixtureApp); len(got) != 0 {
245 t.Errorf("expired result(s) survived the rebuild: %v", got)
246 }
247 // And nothing is left pointing at an app with no work.
248 if d := fx.srv.ScanQueueDepth(); d != 0 {
249 t.Errorf("index depth %d after collecting everything, want 0", d)
250 }
251}
252
253// TestRebuildKeepsAnItemInsideItsTTL is the other half of the test above, and
254// it is here because "collects expired items" passes trivially against a
255// rebuild that deletes everything it finds.
256func TestRebuildKeepsAnItemInsideItsTTL(t *testing.T) {
257 ctx := context.Background()
258 fx, _, _ := drainFixture(t)
259 enableStore(t, fx, fixtureAcct, fixtureApp, true)
260 enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/fresh.pdf", time.Now().Add(-time.Hour))
261
262 if err := fx.srv.rebuildScanQueue(ctx); err != nil {
263 t.Fatal(err)
264 }
265 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
266 t.Fatalf("%d item(s) after the rebuild, want the one that is an hour old", n)
267 }
268 if d := fx.srv.ScanQueueDepth(); d != 1 {
269 t.Errorf("index depth %d, want 1 — a kept item with no index entry is stranded again", d)
270 }
271}
272
273// TestDrainRunsWhileSweepIsBlocked is §7.2 stated as something observable.
274//
275// "TestDrainIsItsOwnGoroutine" as a name would pass against anything at all —
276// against a queue phase inside scanPass, even. So the sweep is PARKED, inside
277// the lister, for the whole duration of a drain, and the drain has to complete
278// anyway. That is the only formulation that fails if the two are ever folded
279// together, which is what draft 1 proposed and what would cost this path the
280// seconds of latency it exists for.
281func TestDrainRunsWhileSweepIsBlocked(t *testing.T) {
282 ctx := context.Background()
283 fx, sc, _ := drainFixture(t)
284 enableStore(t, fx, fixtureAcct, fixtureApp, true)
285 sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/urgent.pdf", "etag-u", time.Now(), 10)
286 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/urgent.pdf", time.Now())
287
288 entered := make(chan struct{}, 1)
289 release := make(chan struct{})
290 fx.srv.storeListerFor = func(context.Context, *objstore.Provisioned) (objstore.ObjectLister, error) {
291 select {
292 case entered <- struct{}{}:
293 default:
294 }
295 <-release
296 return &countingLister{}, nil
297 }
298 swept := make(chan struct{})
299 go func() {
300 _ = fx.srv.scanPass(ctx)
301 close(swept)
302 }()
303
304 select {
305 case <-entered:
306 case <-time.After(5 * time.Second):
307 close(release)
308 t.Fatal("the sweep never reached the lister — the test never parked it")
309 }
310
311 // The sweep is now stopped dead inside scanOneApp. The drain must not be.
312 if err := fx.srv.scanDrainTick(ctx); err != nil {
313 t.Fatalf("drain tick while the sweep is parked: %v", err)
314 }
315 if n := sc.Scans("uploads/urgent.pdf"); n != 1 {
316 t.Fatalf("scanned %d times while the sweep held its budget, want 1", n)
317 }
318
319 close(release)
320 <-swept
321}
322
323// TestDrainRoundRobinFairness: one app's flood must not delay another app's
324// single item beyond one tick.
325//
326// FIFO across the deployment is the design this refuses. The flooder is given
327// more items than the per-app cap on purpose, so that a drain which emptied
328// one app before moving on would still be inside the flooder when the tick
329// ended, and the other app's item would go unanswered.
330func TestDrainRoundRobinFairness(t *testing.T) {
331 ctx := context.Background()
332 fx, sc, _ := drainFixture(t)
333 enableStore(t, fx, fixtureAcct, fixtureApp, true)
334 enableStore(t, fx, fixtureAcct, "flooder", true)
335
336 flood := scanDrainPerApp + 10
337 for i := 0; i < flood; i++ {
338 key := fmt.Sprintf("uploads/flood-%03d.bin", i)
339 sc.PutObject(drainBucket(fixtureAcct, "flooder"), key, "etag-f", time.Now(), 10)
340 enqueueFor(t, fx, fixtureAcct, "flooder", key, time.Now())
341 }
342 sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/urgent.pdf", "etag-u", time.Now(), 10)
343 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/urgent.pdf", time.Now())
344
345 if err := fx.srv.scanDrainTick(ctx); err != nil {
346 t.Fatal(err)
347 }
348 if n := sc.Scans("uploads/urgent.pdf"); n != 1 {
349 t.Fatalf("the urgent item scanned %d times in the tick that also met a flood of %d, want 1", n, flood)
350 }
351 // And the cap held: the flooder did not get the whole queue in one turn.
352 left := queueLen(t, fx, fixtureAcct, "flooder")
353 if left != flood-scanDrainPerApp {
354 t.Errorf("flooder has %d item(s) left, want %d — the per-app cap is not being applied",
355 left, flood-scanDrainPerApp)
356 }
357 // A SECOND tick, because the release rule needs one: DrainApp reports a
358 // consumed page as "not empty" — a full page may have more behind it — so
359 // an app's entry goes on the first tick that lists nothing, not on the
360 // tick that emptied it. After tick two the urgent app is released and the
361 // flooder, which still had a partial page, is not.
362 if err := fx.srv.scanDrainTick(ctx); err != nil {
363 t.Fatal(err)
364 }
365 if n := queueLen(t, fx, fixtureAcct, "flooder"); n != 0 {
366 t.Errorf("flooder has %d item(s) after two ticks, want 0", n)
367 }
368 if d := fx.srv.ScanQueueDepth(); d != 1 {
369 t.Errorf("index depth %d after two ticks, want 1 — the drained app should have been released and the flooder kept", d)
370 }
371}
372
373// TestDrainAgeSweepsResultObjects: per VISITED app, so a console that never
374// restarts does not accumulate them.
375//
376// Nothing else deletes these. The drain removes a result only when a later
377// scan records a real verdict on the same key, and for an object that is
378// permanently untaggable — the ten-tag case — that never happens.
379func TestDrainAgeSweepsResultObjects(t *testing.T) {
380 ctx := context.Background()
381 fx, sc, dir := drainFixture(t)
382 enableStore(t, fx, fixtureAcct, fixtureApp, true)
383 sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/live.pdf", "etag-l", time.Now(), 10)
384 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/live.pdf", time.Now())
385
386 old := time.Now().Add(-objstore.QueueTTL - time.Hour)
387 for _, k := range []string{"uploads/old.pdf", "uploads/recent.pdf"} {
388 if err := objstore.WriteScanResult(ctx, fx.rel, fixtureAcct, fixtureApp, &objstore.ScanResult{
389 Key: k, Verdict: "clean", Version: "etag-x@y", Engine: "clamav", Signature: "sig",
390 At: time.Now().UTC(),
391 }); err != nil {
392 t.Fatal(err)
393 }
394 }
395 oldPath := filepath.Join(dir, filepath.FromSlash(
396 objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/old.pdf")))
397 if err := os.Chtimes(oldPath, old, old); err != nil {
398 t.Fatal(err)
399 }
400
401 if err := fx.srv.scanDrainTick(ctx); err != nil {
402 t.Fatal(err)
403 }
404 got := resultKeys(t, fx, fixtureAcct, fixtureApp)
405 if len(got) != 1 {
406 t.Fatalf("results after the tick = %v, want only the recent one", got)
407 }
408 // The one inside its TTL survives — a collector that dropped both would
409 // pass a test that only counted the expired one.
410 if want := objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/recent.pdf"); got[0] != want {
411 t.Errorf("survivor = %q, want %q", got[0], want)
412 }
413}
414
415// TestQueueDepthReported: §4.4 wants it deployment-wide and §10 wants it per
416// app, and they are different numbers measured in different ways.
417func TestQueueDepthReported(t *testing.T) {
418 ctx := context.Background()
419 fx, _, _ := drainFixture(t)
420 enableStore(t, fx, fixtureAcct, fixtureApp, true)
421 enableStore(t, fx, fixtureAcct, "second", true)
422
423 if d := fx.srv.ScanQueueDepth(); d != 0 {
424 t.Fatalf("depth %d on an idle deployment, want 0", d)
425 }
426 enqueueFor(t, fx, fixtureAcct, fixtureApp, "a.pdf", time.Now())
427 enqueueFor(t, fx, fixtureAcct, fixtureApp, "b.pdf", time.Now())
428 enqueueFor(t, fx, fixtureAcct, "second", "c.pdf", time.Now())
429
430 // Deployment-wide counts APPS with work, not items: two apps, three items.
431 if d := fx.srv.ScanQueueDepth(); d != 2 {
432 t.Errorf("deployment depth = %d, want 2 apps with work", d)
433 }
434 // Per app counts ITEMS.
435 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 2 {
436 t.Errorf("per-app depth = %d, want 2 items", n)
437 }
438 if n := queueLen(t, fx, fixtureAcct, "second"); n != 1 {
439 t.Errorf("per-app depth = %d, want 1 item", n)
440 }
441 // An app that has never enqueued has no prefix, and that is depth zero
442 // rather than an error.
443 if n := queueLen(t, fx, fixtureAcct, "never-used"); n != 0 {
444 t.Errorf("depth %d for an app with no queue prefix, want 0", n)
445 }
446 _ = ctx
447}
448
449// TestDrainDoesNothingWhileTheDeploymentIsDisabled, and specifically KEEPS the
450// items: the deployment switch is the one reversible refusal, and an operator
451// running `enable` should find the queue still there. The TTL is what clears a
452// deployment left off for a day, not the tick.
453func TestDrainDoesNothingWhileTheDeploymentIsDisabled(t *testing.T) {
454 ctx := context.Background()
455 fx, sc, dir := drainFixture(t)
456 counting := &countingStore{DirStore: release.NewDirStore(dir)}
457 fx.rel = counting
458 fx.srv.cfg.Releases = counting
459 enableStore(t, fx, fixtureAcct, fixtureApp, true)
460 sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
461 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
462 setDeploymentScanning(t, fx, false)
463
464 // The tick's own switch check is not what keeps the bytes safe — DrainApp
465 // re-reads consent and refuses on its own, so removing the tick's check
466 // changes no verdict and no item. What it changes is WORK: with it, a
467 // disabled deployment costs one record read; without it, every app with a
468 // queue entry is listed every 30s forever. Listings are therefore what
469 // this asserts, because they are the only observable difference.
470 counting.lists.Store(0)
471 if err := fx.srv.scanDrainTick(ctx); err != nil {
472 t.Fatal(err)
473 }
474 if n := counting.lists.Load(); n != 0 {
475 t.Errorf("%d listing(s) on a disabled deployment — the tick is doing per-app work it can never use", n)
476 }
477 if n := sc.Scans("uploads/x.pdf"); n != 0 {
478 t.Errorf("scanned %d times with the deployment switch off", n)
479 }
480 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
481 t.Fatalf("%d item(s) left, want the item KEPT — this refusal is reversible", n)
482 }
483
484 // And it drains once the operator turns it back on, without a restart.
485 setDeploymentScanning(t, fx, true)
486 if err := fx.srv.scanDrainTick(ctx); err != nil {
487 t.Fatal(err)
488 }
489 if n := sc.Scans("uploads/x.pdf"); n != 1 {
490 t.Errorf("scanned %d times after re-enabling, want 1", n)
491 }
492}
493
494// TestDrainDropsTheQueueWhenConsentIsGone is §7.5's other side: the permanent
495// refusals delete, because none of them can change back in a way that makes
496// the item useful, and an item that outlives consent is a request to read
497// bytes nobody may read.
498func TestDrainDropsTheQueueWhenConsentIsGone(t *testing.T) {
499 ctx := context.Background()
500 fx, sc, _ := drainFixture(t)
501 enableStore(t, fx, fixtureAcct, fixtureApp, true)
502 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
503 // The operator revokes the per-store grant between enqueue and drain.
504 revokeScanning(t, fx, fixtureAcct, fixtureApp)
505
506 if err := fx.srv.scanDrainTick(ctx); err != nil {
507 t.Fatal(err)
508 }
509 if n := sc.Scans("uploads/x.pdf"); n != 0 {
510 t.Errorf("read bytes %d time(s) from a store that may not be scanned", n)
511 }
512 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
513 t.Errorf("%d item(s) survived a permanent refusal", n)
514 }
515}
516
517// TestScanNotifyWakesAWaiter is the long-poll's half of the drain (§5.2). The
518// door lands in task 13; the wake it selects on is delivered here, so it has
519// something to be written against.
520func TestScanNotifyWakesAWaiter(t *testing.T) {
521 ctx := context.Background()
522 fx, sc, _ := drainFixture(t)
523 enableStore(t, fx, fixtureAcct, fixtureApp, true)
524 sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
525 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
526
527 ref := objstore.AppRef{Account: fixtureAcct, App: fixtureApp}
528 woken, cancel, ok := fx.srv.scanWait(ref, "uploads/x.pdf")
529 if !ok {
530 t.Fatal("an idle console refused to park the first waiter")
531 }
532 defer cancel()
533 // A waiter on a DIFFERENT key must not be woken by this drain.
534 other, cancelOther, ok := fx.srv.scanWait(ref, "uploads/other.pdf")
535 if !ok {
536 t.Fatal("an idle console refused to park a second waiter")
537 }
538 defer cancelOther()
539
540 if err := fx.srv.scanDrainTick(ctx); err != nil {
541 t.Fatal(err)
542 }
543 select {
544 case <-woken:
545 default:
546 t.Error("the drain scanned the object and nothing woke the waiter")
547 }
548 select {
549 case <-other:
550 t.Error("a waiter on another key was woken")
551 default:
552 }
553
554 // Cancelling removes the registration rather than leaking one entry per
555 // key ever waited on.
556 cancel()
557 fx.srv.scanWaits.mu.Lock()
558 n := len(fx.srv.scanWaits.m[scanWaitKey{ref: ref, key: "uploads/x.pdf"}])
559 fx.srv.scanWaits.mu.Unlock()
560 if n != 0 {
561 t.Errorf("%d waiter(s) still registered after cancel", n)
562 }
563}
564
565// TestDrainNotifiesEvenWhenALaterItemFails: DrainApp returns what it managed
566// before it stopped, and a waiter whose object WAS scanned must be woken even
567// though the app's next item failed. Notifying after the error check would
568// make a long-poll wait out its whole window for an answer that already
569// exists on the object.
570//
571// Reaching that state takes some care with the fake. The FIRST object is
572// pre-tagged at its current version, so Status answers from the tag and Scan
573// is never called for it — which leaves the one-shot Scan failure below for
574// the SECOND object. done therefore carries the first while err carries the
575// second, which is exactly the shape being asserted.
576func TestDrainNotifiesEvenWhenALaterItemFails(t *testing.T) {
577 ctx := context.Background()
578 fx, sc, _ := drainFixture(t)
579 enableStore(t, fx, fixtureAcct, fixtureApp, true)
580 bucket := drainBucket(fixtureAcct, fixtureApp)
581
582 mod := time.Now().Truncate(time.Second)
583 first := objstore.ScanObject{Key: "uploads/first.pdf", ETag: "etag-1", Modified: mod, Size: 10}
584 sc.PutObject(bucket, first.Key, first.ETag, mod, first.Size)
585 sc.Put(bucket, first.Key, first.Version(), objstore.VerdictClean)
586 sc.PutObject(bucket, "uploads/second.pdf", "etag-2", mod, 10)
587
588 // Ids sort by time, so the first item is drained first.
589 enqueueFor(t, fx, fixtureAcct, fixtureApp, first.Key, time.Now())
590 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/second.pdf", time.Now().Add(2*time.Millisecond))
591
592 ref := objstore.AppRef{Account: fixtureAcct, App: fixtureApp}
593 woken, cancel, ok := fx.srv.scanWait(ref, first.Key)
594 if !ok {
595 t.Fatal("an idle console refused to park a waiter")
596 }
597 defer cancel()
598
599 sc.Fail("Scan", errors.New("engine had a bad moment"))
600
601 if err := fx.srv.scanDrainTick(ctx); err != nil {
602 t.Fatal(err) // the tick logs per-app failures rather than returning them
603 }
604 if sc.Scans(first.Key) != 0 {
605 t.Fatal("the pre-tagged object was rescanned, so the Scan failure did not reach the second item")
606 }
607 select {
608 case <-woken:
609 default:
610 t.Error("the first object was answered and its waiter was not woken, because a later item failed")
611 }
612 // The failed item is KEPT — an engine failure is about the engine, not
613 // about whether this request was legitimate.
614 if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
615 t.Errorf("%d item(s) left, want the failed one kept for the next tick", n)
616 }
617}
618
619// TestStartScanDrainRebuildsBeforeItsFirstTick covers the goroutine itself,
620// which the tick-level tests above do not: they call rebuildScanQueue and
621// scanDrainTick by hand, so a StartScanDrain that never ran the rebuild would
622// pass every one of them and strand every item across every restart.
623//
624// It is also the only test of "the first tick is immediate". A loop that
625// waited a full interval before its first pass would make every restart cost
626// 30s of latency on work that was already queued.
627func TestStartScanDrainRebuildsBeforeItsFirstTick(t *testing.T) {
628 fx, sc, _ := drainFixture(t)
629 enableStore(t, fx, fixtureAcct, fixtureApp, true)
630 sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
631 enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
632
633 ctx, cancel := context.WithCancel(context.Background())
634 defer cancel()
635 stopped := make(chan struct{})
636 go func() {
637 fx.srv.StartScanDrain(ctx)
638 close(stopped)
639 }()
640
641 deadline := time.Now().Add(5 * time.Second)
642 for sc.Scans("uploads/x.pdf") == 0 {
643 if time.Now().After(deadline) {
644 cancel()
645 <-stopped
646 t.Fatal("the goroutine did not rebuild and drain a stranded item within 5s — well inside its first tick")
647 }
648 time.Sleep(2 * time.Millisecond)
649 }
650 cancel()
651 select {
652 case <-stopped:
653 case <-time.After(5 * time.Second):
654 t.Fatal("StartScanDrain did not return when its context was cancelled")
655 }
656}
657
658// TestStartScanDrainIsInertWithoutAnEngine: scanning is opt-in, and the
659// ordinary deployment has no clamd. The goroutine must return rather than tick
660// forever over a deployment that can never drain anything.
661func TestStartScanDrainIsInertWithoutAnEngine(t *testing.T) {
662 fx, _, _ := drainFixture(t)
663 fx.srv.cfg.StoreScanner = nil
664
665 stopped := make(chan struct{})
666 go func() {
667 fx.srv.StartScanDrain(context.Background())
668 close(stopped)
669 }()
670 select {
671 case <-stopped:
672 case <-time.After(5 * time.Second):
673 t.Fatal("StartScanDrain kept ticking on a deployment with no engine")
674 }
675}
676
677func TestScanUsageCountsOnDemandOnly(t *testing.T) {
678 ctx := context.Background()
679 fx, sc, _ := drainFixture(t)
680 enableStore(t, fx, fixtureAcct, fixtureApp, true)
681 bucket := drainBucket(fixtureAcct, fixtureApp)
682 mod := time.Now().Truncate(time.Second)
683
684 tagged := objstore.ScanObject{Key: "uploads/tagged.pdf", ETag: "etag-tagged", Modified: mod, Size: 100}
685 sc.PutObject(bucket, tagged.Key, tagged.ETag, tagged.Modified, tagged.Size)
686 sc.Put(bucket, tagged.Key, tagged.Version(), objstore.VerdictClean)
687 sc.PutObject(bucket, "uploads/new.pdf", "etag-new", mod, 25)
688 enqueueFor(t, fx, fixtureAcct, fixtureApp, tagged.Key, mod)
689 enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/new.pdf", mod.Add(time.Millisecond))
690
691 if err := fx.srv.scanDrainTick(ctx); err != nil {
692 t.Fatal(err)
693 }
694 raw, _, err := fx.rel.Get(ctx, usage.ScanDayKey(usage.DayOf(time.Now())))
695 if err != nil {
696 t.Fatal(err)
697 }
698 var day usage.ScanDay
699 if err := json.Unmarshal(raw, &day); err != nil {
700 t.Fatal(err)
701 }
702 if got := day.Accounts[fixtureAcct]; got != (usage.ScanRow{Objects: 1, Bytes: 25}) {
703 t.Fatalf("usage = %+v, want only the newly scanned object", got)
704 }
705}
706
707func TestScanUsageFoldsPerTick(t *testing.T) {
708 ctx := context.Background()
709 fx, sc, dir := drainFixture(t)
710 counting := &scanUsageCountingStore{DirStore: release.NewDirStore(dir)}
711 fx.rel = counting
712 fx.srv.cfg.Releases = counting
713 enableStore(t, fx, fixtureAcct, fixtureApp, true)
714 enableStore(t, fx, fixtureAcct, "second", true)
715 for app, key := range map[string]string{fixtureApp: "uploads/a.pdf", "second": "uploads/b.pdf"} {
716 sc.PutObject(drainBucket(fixtureAcct, app), key, "etag-1", time.Now(), 10)
717 enqueueFor(t, fx, fixtureAcct, app, key, time.Now())
718 }
719
720 if err := fx.srv.scanDrainTick(ctx); err != nil {
721 t.Fatal(err)
722 }
723 if got := counting.puts.Load(); got != 1 {
724 t.Fatalf("usage writes = %d, want one for the whole tick", got)
725 }
726 raw, _, err := counting.Get(ctx, usage.ScanDayKey(usage.DayOf(time.Now())))
727 if err != nil {
728 t.Fatal(err)
729 }
730 var day usage.ScanDay
731 if err := json.Unmarshal(raw, &day); err != nil {
732 t.Fatal(err)
733 }
734 if got := day.Accounts[fixtureAcct]; got != (usage.ScanRow{Objects: 2, Bytes: 20}) {
735 t.Fatalf("folded usage = %+v", got)
736 }
737}
738