| 1 | // The scan drain (spec §7.2): the goroutine that empties the on-demand queue. |
| 2 | // |
| 3 | // ITS OWN GOROUTINE AND ITS OWN CLOCK, which is the whole reason this file is |
| 4 | // not three functions inside scanpass.go. Draft 1 put a queue phase inside the |
| 5 | // existing pass; that cannot deliver seconds of latency, because scanPass |
| 6 | // sweeps apps sequentially and each gets a full budget (scanpass.go), so a |
| 7 | // queue phase behind it starves for up to N × budget. An app that uploads a |
| 8 | // file and waits for a verdict would be waiting on a walk of somebody else's |
| 9 | // bucket. |
| 10 | // |
| 11 | // The only state it shares with the sweep is object tags, which are idempotent |
| 12 | // per version: two writers agreeing on the same answer for the same bytes is |
| 13 | // not a race. It never touches the sidecar. |
| 14 | package console |
| 15 | |
| 16 | import ( |
| 17 | "context" |
| 18 | "errors" |
| 19 | "log" |
| 20 | "strings" |
| 21 | "sync" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/carlosframework/platform/internal/objstore" |
| 25 | "github.com/carlosframework/platform/internal/usage" |
| 26 | ) |
| 27 | |
| 28 | // scanDrainInterval is the tick. 30s, against the sweep's hour: this path |
| 29 | // exists because waiting an hour for a verdict on an upload is useless. |
| 30 | const scanDrainInterval = 30 * time.Second |
| 31 | |
| 32 | // scanDrainPerApp is §7.3's fairness cap — how many items ONE app gets in one |
| 33 | // tick, not how many the deployment gets. |
| 34 | // |
| 35 | // Each app gets a bounded turn; excess requests stay queued for later ticks. |
| 36 | const scanDrainPerApp = 50 |
| 37 | |
| 38 | // scanDrainAppBudget bounds one app's turn in one tick — the time half of the |
| 39 | // fairness rule, where scanDrainPerApp is the count half. |
| 40 | // |
| 41 | // Longer than a tick on purpose: a drain that overran its interval should |
| 42 | // finish the objects it started rather than be cut off mid-scan, and the |
| 43 | // ticker's next fire simply waits. What it must not do is run unbounded. |
| 44 | const scanDrainAppBudget = 90 * time.Second |
| 45 | |
| 46 | // StartScanDrain rebuilds the index and then drains every 30s until ctx is |
| 47 | // done. serve() starts it; constructing a Server starts nothing. |
| 48 | // |
| 49 | // THE REBUILD RUNS FIRST AND BLOCKS THE FIRST TICK, because a tick before it |
| 50 | // would drain only the apps this process happens to have heard from — which, |
| 51 | // one second after a restart, is none of them. |
| 52 | func (s *Server) StartScanDrain(ctx context.Context) { |
| 53 | if s.cfg.Releases == nil || s.cfg.StoreScanner == nil { |
| 54 | return // no bucket, or no engine: nothing to drain into |
| 55 | } |
| 56 | s.scanDelivery = newScanDeliveryPool(ctx, s) |
| 57 | defer func() { |
| 58 | s.scanDelivery.stop() |
| 59 | s.scanDelivery = nil |
| 60 | }() |
| 61 | rebuilt := true |
| 62 | if err := s.rebuildScanQueue(ctx); err != nil { |
| 63 | rebuilt = false |
| 64 | // Not fatal. A console that cannot rebuild still drains what it is |
| 65 | // told about from here on, and the next restart tries again; refusing |
| 66 | // to start the goroutine would turn a listing hiccup into an outage |
| 67 | // of the whole on-demand path. |
| 68 | log.Printf("console: scan queue rebuild: %v", err) |
| 69 | } |
| 70 | rebuildDelay := scanDrainInterval |
| 71 | nextRebuild := time.Now().Add(rebuildDelay) |
| 72 | t := time.NewTicker(scanDrainInterval) |
| 73 | defer t.Stop() |
| 74 | for { |
| 75 | // Retry failed discovery with backoff. An app held by another owner |
| 76 | // must not cause a deployment-wide enumeration on every drain tick. |
| 77 | if !rebuilt && !time.Now().Before(nextRebuild) { |
| 78 | if err := s.rebuildScanQueue(ctx); err != nil { |
| 79 | log.Printf("console: scan queue rebuild (retry): %v", err) |
| 80 | rebuildDelay = min(rebuildDelay*2, 5*time.Minute) |
| 81 | nextRebuild = time.Now().Add(rebuildDelay) |
| 82 | } else { |
| 83 | rebuilt = true |
| 84 | } |
| 85 | } |
| 86 | if err := s.scanDrainTick(ctx); err != nil { |
| 87 | log.Printf("console: scan drain: %v", err) |
| 88 | } |
| 89 | select { |
| 90 | case <-ctx.Done(): |
| 91 | return |
| 92 | case <-t.C: |
| 93 | } |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | // rebuildScanQueue is the one full enumeration (§7.1): every app on the |
| 98 | // deployment, so that an item stranded by a crash between the object write and |
| 99 | // the index bump is found, and everything past its TTL is collected. |
| 100 | // |
| 101 | // Same enumeration as scanPass, and the same accepted cost — but ONCE at boot |
| 102 | // rather than hourly, so it is the cheap one of the two. |
| 103 | // |
| 104 | // One app's failure does not stop the others: this pass exists to recover what |
| 105 | // it can, and an account whose listing fails should not cost every account |
| 106 | // after it its queue. |
| 107 | func (s *Server) rebuildScanQueue(ctx context.Context) error { |
| 108 | if s.cfg.Releases == nil { |
| 109 | return nil |
| 110 | } |
| 111 | now := time.Now() |
| 112 | acctPrefixes, err := s.cfg.Releases.ListDirs(ctx, "apps/") |
| 113 | if err != nil { |
| 114 | return err |
| 115 | } |
| 116 | var errs []error |
| 117 | for _, acctPrefix := range acctPrefixes { |
| 118 | acct := strings.TrimSuffix(strings.TrimPrefix(acctPrefix, "apps/"), "/") |
| 119 | appPrefixes, err := s.cfg.Releases.ListDirs(ctx, acctPrefix) |
| 120 | if err != nil { |
| 121 | log.Printf("console: scan queue rebuild: %s: %v", acct, err) |
| 122 | errs = append(errs, err) |
| 123 | continue |
| 124 | } |
| 125 | for _, appPrefix := range appPrefixes { |
| 126 | if err := ctx.Err(); err != nil { |
| 127 | return err |
| 128 | } |
| 129 | app := strings.TrimSuffix(strings.TrimPrefix(appPrefix, acctPrefix), "/") |
| 130 | if _, err := objstore.RebuildQueue(ctx, s.cfg.Releases, s.scanQueue, acct, app, now); err != nil { |
| 131 | log.Printf("console: scan queue rebuild: %s/%s: %v", acct, app, err) |
| 132 | errs = append(errs, err) |
| 133 | } |
| 134 | // The fallback results are collected in the SAME pass, because |
| 135 | // this is the only place that visits an app which has stopped |
| 136 | // enqueuing. The per-tick collector below only reaches apps that |
| 137 | // still have work. |
| 138 | s.collectScanResults(ctx, acct, app, now) |
| 139 | } |
| 140 | } |
| 141 | // Joined, not swallowed: StartScanDrain retries until this returns nil, so |
| 142 | // a partial enumeration must not look like a complete one. |
| 143 | return errors.Join(errs...) |
| 144 | } |
| 145 | |
| 146 | // scanDrainTick gives every app with work one capped turn. |
| 147 | // |
| 148 | // Round-robin is what iterating the index IS: every app named in it gets a |
| 149 | // turn each tick and none gets more than scanDrainPerApp, so an app cannot |
| 150 | // buy latency from another by queueing more. |
| 151 | // |
| 152 | // The deployment switch is read every tick, exactly as scanPass reads it, so |
| 153 | // `carlos system antivirus enable` takes effect without a restart. Disabled |
| 154 | // means this tick does nothing — not that there is no goroutine, which is what |
| 155 | // would make enabling need one. |
| 156 | func (s *Server) scanDrainTick(ctx context.Context) error { |
| 157 | // An early-out for a direct caller, not a guarantee: StartScanDrain never |
| 158 | // starts without an engine, and DrainApp refuses with ErrNoEngine before |
| 159 | // it lists anything. Removing this line changes nothing observable — it |
| 160 | // only stops a tick from logging one refusal per app. Kept because |
| 161 | // scanPass guards the same way, and said here so nobody looks for the |
| 162 | // test that pins it. |
| 163 | if s.cfg.Releases == nil || s.cfg.StoreScanner == nil { |
| 164 | return nil |
| 165 | } |
| 166 | enabled, err := s.antivirusEnabled(ctx) |
| 167 | if err != nil { |
| 168 | return err |
| 169 | } |
| 170 | if !enabled { |
| 171 | // NOT THE SAFETY NET, and worth being exact about: DrainApp re-reads |
| 172 | // consent for itself and refuses a disabled deployment on its own, so |
| 173 | // deleting this check changes no verdict and loses no item. What it |
| 174 | // changes is WORK — with it a disabled deployment costs one record |
| 175 | // read per tick, without it every app with an index entry is listed |
| 176 | // every 30s for as long as it stays off. |
| 177 | // |
| 178 | // Items are KEPT, not dropped: this refusal is reversible, and the |
| 179 | // TTL is what eventually clears a deployment left off for a day. |
| 180 | return nil |
| 181 | } |
| 182 | now := time.Now() |
| 183 | deltas := map[string]usage.ScanRow{} |
| 184 | for _, ref := range s.scanQueue.Apps() { |
| 185 | if err := ctx.Err(); err != nil { |
| 186 | return err |
| 187 | } |
| 188 | delta := s.drainScanApp(ctx, ref, now) |
| 189 | if delta.Objects != 0 { |
| 190 | row := deltas[ref.Account] |
| 191 | row.Objects += delta.Objects |
| 192 | row.Bytes += delta.Bytes |
| 193 | deltas[ref.Account] = row |
| 194 | } |
| 195 | } |
| 196 | return usage.UpsertScanDay(ctx, s.cfg.Releases, now, deltas) |
| 197 | } |
| 198 | |
| 199 | // drainScanApp drains one app and tells whoever is waiting. |
| 200 | // |
| 201 | // Every outcome is logged and none is returned: one app's engine trouble must |
| 202 | // not end the tick for the apps after it in the round-robin, which is the |
| 203 | // posture scanPass already takes for the same reason. |
| 204 | func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time.Time) usage.ScanRow { |
| 205 | var delta usage.ScanRow |
| 206 | parent := ctx |
| 207 | // ONE APP'S TURN IS BOUNDED. clamd's own read timeout is minutes, and the |
| 208 | // object body stream carries no deadline of its own, so fifty slow objects |
| 209 | // belonging to one tenant could hold the whole deployment's drain for |
| 210 | // hours — starving every other tenant's single urgent item, which is the |
| 211 | // precise failure §7.3's cap exists to prevent and which the cap alone |
| 212 | // does not close. The sweep budgets per app for the same reason; this is |
| 213 | // that budget. An app that runs out keeps its items and its index entry |
| 214 | // and gets the next tick. |
| 215 | ctx, cancel := context.WithTimeout(ctx, scanDrainAppBudget) |
| 216 | defer cancel() |
| 217 | |
| 218 | startRec, _, _ := objstore.Get(ctx, s.cfg.Releases, ref.Account, ref.App) |
| 219 | done, _, err := objstore.DrainApp(ctx, s.cfg.Releases, s.cfg.StoreScanner, s.scanQueue, |
| 220 | ref.Account, ref.App, scanDrainPerApp, now) |
| 221 | |
| 222 | // NOTIFY BEFORE HANDLING THE ERROR. DrainApp returns the items it managed |
| 223 | // before it stopped, and a waiter whose object was actually scanned should |
| 224 | // be woken even though the app's next item failed. Dropping these on the |
| 225 | // error path would make a long-poll wait out its full window for an answer |
| 226 | // that already exists. |
| 227 | for _, d := range done { |
| 228 | if d.Scanned { |
| 229 | delta.Objects++ |
| 230 | delta.Bytes += d.Size |
| 231 | } |
| 232 | s.scanNotify(ref, d.Key) |
| 233 | for _, path := range d.CallbackPaths { |
| 234 | if startRec != nil && startRec.Scanning != nil { |
| 235 | s.scanDelivery.enqueue(scanDeliveryJob{ref: ref, done: d, path: path, gen: startRec.Scanning.WebhookGen}) |
| 236 | } |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | switch { |
| 241 | case err == nil: |
| 242 | case errors.Is(err, objstore.ErrQueueNotPermitted): |
| 243 | // Not an incident: consent was withdrawn, or the app was trashed, and |
| 244 | // DrainApp has already deleted the items and the results. Logged at |
| 245 | // all only because "the queue emptied without scanning" is a thing an |
| 246 | // operator reading a support question wants to be able to see. |
| 247 | log.Printf("console: scan drain: %s/%s: queue dropped — this store may no longer be scanned", |
| 248 | ref.Account, ref.App) |
| 249 | case errors.Is(err, objstore.ErrScanningDisabled): |
| 250 | // A disable that landed between the tick's check and this app. The |
| 251 | // items stay; the next enabled tick has them. |
| 252 | default: |
| 253 | log.Printf("console: scan drain: %s/%s: %v", ref.Account, ref.App, err) |
| 254 | } |
| 255 | cleanupCtx, cleanupCancel := context.WithTimeout(parent, storeTimeout) |
| 256 | defer cleanupCancel() |
| 257 | s.collectScanResults(cleanupCtx, ref.Account, ref.App, now) |
| 258 | return delta |
| 259 | } |
| 260 | |
| 261 | // collectScanResults ages out an app's fallback result objects. |
| 262 | // |
| 263 | // PER VISITED APP rather than in one deployment-wide sweep, so that a console |
| 264 | // which never restarts does not accumulate them — nothing else deletes a |
| 265 | // result for an object that stays permanently untaggable, because the drain |
| 266 | // only removes one when a later scan records a real verdict on the same key. |
| 267 | func (s *Server) collectScanResults(ctx context.Context, acct, app string, now time.Time) { |
| 268 | n, err := objstore.CollectScanResults(ctx, s.cfg.Releases, acct, app, now.Add(-objstore.QueueTTL)) |
| 269 | switch { |
| 270 | case errors.Is(err, objstore.ErrNoModLister): |
| 271 | // A store that cannot report modification times, which is a test fake |
| 272 | // rather than a deployment. Skipped, not failed — storagePass treats |
| 273 | // its own optional interface the same way. |
| 274 | case err != nil: |
| 275 | log.Printf("console: scan result collect: %s/%s: %v", acct, app, err) |
| 276 | case n > 0: |
| 277 | log.Printf("console: scan result collect: %s/%s: dropped %d expired result(s)", acct, app, n) |
| 278 | } |
| 279 | } |
| 280 | |
| 281 | // ScanQueueDepth is how many apps have on-demand work waiting — the |
| 282 | // deployment-wide number §4.4 wants on `carlos system antivirus status`. |
| 283 | // |
| 284 | // APPS, not items, and named for what it counts. The per-app item count is |
| 285 | // objstore.QueueDepth, which costs a listing; this one is free and is the |
| 286 | // number that answers "is the drain keeping up". |
| 287 | func (s *Server) ScanQueueDepth() int { |
| 288 | if s.scanQueue == nil { |
| 289 | return 0 |
| 290 | } |
| 291 | return s.scanQueue.Len() |
| 292 | } |
| 293 | |
| 294 | // ---- waking the long-poll --------------------------------------------------- |
| 295 | |
| 296 | // scanWaitKey is one object, for one app. |
| 297 | type scanWaitKey struct { |
| 298 | ref objstore.AppRef |
| 299 | key string |
| 300 | } |
| 301 | |
| 302 | // scanWaiters is the in-process notification the status door's `wait` uses |
| 303 | // (§5.2), so a long-poll costs a channel rather than a tag read per second per |
| 304 | // waiter. |
| 305 | // |
| 306 | // IN-PROCESS ONLY, and that is a correctness statement rather than a caveat: a |
| 307 | // waiter on console A is never woken by a drain on console B. The door must |
| 308 | // therefore treat a wake as an OPTIMISATION and still read the tag when its |
| 309 | // timer fires — never as the only way an answer arrives. |
| 310 | type scanWaiters struct { |
| 311 | mu sync.Mutex |
| 312 | m map[scanWaitKey]map[chan struct{}]struct{} |
| 313 | // perApp and total are the two caps (§5.2). Counted rather than derived |
| 314 | // from m, because m is keyed by OBJECT: one app parking a thousand |
| 315 | // requests across a thousand keys is a thousand map entries of one, which |
| 316 | // a per-key length would read as no load at all. |
| 317 | perApp map[objstore.AppRef]int |
| 318 | total int |
| 319 | } |
| 320 | |
| 321 | // scanWait registers interest in one object and returns the channel to select |
| 322 | // on, plus the cancel that must be deferred. |
| 323 | // |
| 324 | // The channel is BUFFERED to depth one and the notifier never blocks on it, so |
| 325 | // a waiter that has already given up cannot stall the drain — the goroutine |
| 326 | // whose latency this whole path exists to protect. |
| 327 | func (s *Server) scanWait(ref objstore.AppRef, key string) (ch <-chan struct{}, release func(), ok bool) { |
| 328 | w := &s.scanWaits |
| 329 | k := scanWaitKey{ref: ref, key: key} |
| 330 | c := make(chan struct{}, 1) |
| 331 | w.mu.Lock() |
| 332 | defer w.mu.Unlock() |
| 333 | if w.total >= scanWaitersGlobal || w.perApp[ref] >= scanWaitersPerApp { |
| 334 | // Refused a PARK, not refused an answer. The caller reads the object |
| 335 | // and replies truthfully; only the waiting is skipped, so a full |
| 336 | // console degrades to polling rather than to errors. |
| 337 | return nil, func() {}, false |
| 338 | } |
| 339 | if w.m == nil { |
| 340 | w.m = map[scanWaitKey]map[chan struct{}]struct{}{} |
| 341 | w.perApp = map[objstore.AppRef]int{} |
| 342 | } |
| 343 | if w.m[k] == nil { |
| 344 | w.m[k] = map[chan struct{}]struct{}{} |
| 345 | } |
| 346 | w.m[k][c] = struct{}{} |
| 347 | w.perApp[ref]++ |
| 348 | w.total++ |
| 349 | released := false |
| 350 | return c, func() { |
| 351 | w.mu.Lock() |
| 352 | defer w.mu.Unlock() |
| 353 | // IDEMPOTENT. The status door defers this and the test calls it early; |
| 354 | // a second call that decremented again would leak capacity downwards |
| 355 | // until the console stopped parking anything at all. |
| 356 | if released { |
| 357 | return |
| 358 | } |
| 359 | released = true |
| 360 | if set := w.m[k]; set != nil { |
| 361 | delete(set, c) |
| 362 | if len(set) == 0 { |
| 363 | // The map must not grow one entry per key ever waited on. |
| 364 | delete(w.m, k) |
| 365 | } |
| 366 | } |
| 367 | w.perApp[ref]-- |
| 368 | if w.perApp[ref] <= 0 { |
| 369 | delete(w.perApp, ref) |
| 370 | } |
| 371 | w.total-- |
| 372 | }, true |
| 373 | } |
| 374 | |
| 375 | // scanNotify wakes everyone waiting on one object. Never blocks. |
| 376 | func (s *Server) scanNotify(ref objstore.AppRef, key string) { |
| 377 | w := &s.scanWaits |
| 378 | w.mu.Lock() |
| 379 | defer w.mu.Unlock() |
| 380 | for ch := range w.m[scanWaitKey{ref: ref, key: key}] { |
| 381 | select { |
| 382 | case ch <- struct{}{}: |
| 383 | default: // already woken and not yet read; one wake is all it needs |
| 384 | } |
| 385 | } |
| 386 | } |
| 387 | |
amadan