From c21813cc0784059d59955a3dc5ce23a5c758d079 Mon Sep 17 00:00:00 2001 From: Justin Erenkrantz Date: Tue, 11 Aug 2026 03:21:11 +0000 Subject: [PATCH] Say when the upload queue is filling, not only after it has lost something MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A put writes the local body synchronously and queues the upload, and a submission that finds the queue full is dropped. That is the right trade for a best-effort tier, but the only evidence of it was a counter that moves after the entries are already gone, and the loss surfaces on another machine much later as an ordinary miss. Report the backlog while it is still whole: - Cache.UploadQueue reports depth and capacity, carried on the daemon's status report and exposed as plaid_cache_upload_queue_depth and plaid_cache_upload_queue_capacity. status prints the same pair. - A drop logs a warning naming how many went with it, at most once per 30 seconds. A saturated queue drops thousands a second, so a line each would bury the log; the count since the last report travels in the line instead. - PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH replaces the hard-coded per-worker backlog, so trading memory for burst tolerance is not a code change. - PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT lets a put wait a bounded time for room instead of dropping. Off by default: a put that waits is a build that waits, which is what the drop exists to prevent. Shutdown does not wait it out either — close releases a waiting submit rather than holding exit for the rest of an operator's timeout. Defaults are unchanged: the same 64 slots per worker, and no waiting. Co-authored-by: c1-squire-dev[bot] --- README.md | 16 +++ cmd/plaid-cache/commands.go | 15 ++- internal/cache/cache.go | 124 +++++++++++++++++++-- internal/cache/fake_test.go | 52 ++++++++- internal/cache/upload_test.go | 180 +++++++++++++++++++++++++++++++ internal/config/config.go | 35 ++++++ internal/config/config_test.go | 15 +++ internal/daemon/metrics.go | 14 +++ internal/daemon/metrics_test.go | 10 ++ internal/daemon/protocol.go | 10 ++ internal/daemon/server.go | 1 + internal/daemon/stats_op_test.go | 9 ++ 12 files changed, 469 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 9fe911e..7673e44 100644 --- a/README.md +++ b/README.md @@ -249,6 +249,7 @@ There is no `directory`, no `config`, and no `volume` line, and `remote` says wh | `plaid_cache_gets_total{result}` | counter | Lookups by outcome: `local_hit`, `remote_hit`, `miss`. | | `plaid_cache_puts_total`, `plaid_cache_repairs_total`, `plaid_cache_compactions_total` | counter | Stores, index entries dropped for a missing body, and compactions. | | `plaid_cache_uploads_total{result}` | counter | Uploads by outcome: `ok`, `failed`, `dropped`, `skipped`. | +| `plaid_cache_upload_queue_depth`, `plaid_cache_upload_queue_capacity` | gauge | Uploads waiting on this daemon's pool, and how many may wait. Depth at capacity means uploads are being dropped. | | `plaid_cache_activity_start_time_seconds` | gauge | When the counters above started counting. | Two labels, both with a short fixed set of values. Nothing is ever labelled by digest, key, path, or client: a series is created for every distinct label value and never forgotten, so a per-request label is an unbounded leak in whatever scrapes it. @@ -525,6 +526,8 @@ be a surprising amount of reach for this one to have. | `PLAID_GOCACHE_S3_PREFIX` | Key prefix within the bucket. | empty | | `PLAID_GOCACHE_MIN_UPLOAD_SIZE` | Skip uploading bodies smaller than this. Skipping also omits the action record, so the entry becomes a remote miss and the action is re-run rather than re-downloaded. | `0` (upload everything) | | `PLAID_GOCACHE_UPLOAD_CONCURRENCY` | Remote upload workers. | `NumCPU` | +| `PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH` | Uploads that may wait per worker before further ones are dropped. Raising it buys tolerance of a burst with memory. | `64` | +| `PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT` | How long a put waits for room in the upload queue before dropping the upload, as a Go duration. Zero never waits. | `0` (drop rather than wait) | | `PLAID_GOCACHE_TOUCH_GRANULARITY` | Relatime-style window for last-used updates. | `1h` | | `PLAID_GOCACHE_IDLE_TIMEOUT` | Daemon exits after this long with no connections. | `30m` | | `PLAID_GOCACHE_EVICT_INTERVAL` | Eviction ticker period. | `1m` | @@ -553,6 +556,19 @@ Credentials come from the standard AWS configuration chain. The caller needs `s3 Uploads are best-effort. They run in a bounded worker pool off the critical path, and a failure is logged and counted rather than propagated — a cache must never fail a build. If the daemon cannot be reached at all, the plugin falls back to direct mode, warns on stderr, and lets the build proceed. +#### When the upload queue fills + +The pool is fed by a bounded queue, and a build that produces entries faster than the link ships them fills it. A submission that finds it full is dropped: the body is already on local disk and the put has already returned, so nothing fails and nothing waits. That is the right trade for a tier a build must never stall on, and it is worth knowing what it costs, because the cost is paid on a different machine at a different time. The entry is simply not in the shared tier, so the next reader takes an ordinary miss and rebuilds — and there is nothing about that miss to connect it back to the queue that dropped it. + +So the loss is reported where it happens rather than only where it lands: + +- `plaid_cache_upload_queue_depth` against `plaid_cache_upload_queue_capacity` says how close the queue is to full. This is the number to watch and to alert on: it climbs before anything is lost, where `plaid_cache_uploads_total{result="dropped"}` only moves once entries are already gone. `plaid-cache status` prints the same pair as its `upload q` line. +- The first drop logs a warning naming how many uploads went with it, and no more than one such line is written per 30 seconds — a saturated queue drops thousands of uploads a second, and a line each would bury the log rather than explain it. Each line carries the count since the previous one, so a burst and an hour of sustained shedding do not look alike. It travels with the daemon's other cache diagnostics, so `PLAID_GOCACHE_LOG=info` is what puts it in the log; the two gauges above need nothing turned on. + +Two settings adjust the trade. `PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH` is the backlog each worker may accumulate; raising it absorbs a longer burst in exchange for memory, since the queue holds a job per waiting entry. `PLAID_GOCACHE_UPLOAD_CONCURRENCY` drains it faster, which is the better answer when the bucket, and not the burst, is the constraint. + +`PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT` reverses the trade rather than adjusting it: a put whose queue is full waits up to that long for room and only then drops. It is off by default and should stay off for ordinary builds — it makes a saturated queue something a build can wait on, which is exactly what the drop exists to prevent. It is there for the runs where the upload is the point and the local copy is not, such as a warmer filling a bucket for everyone else, and the wait is bounded so that opting in cannot turn into an unbounded stall. Shutdown does not wait it out either: a put still waiting when the daemon stops drops its upload and lets the process exit. + Keys are laid out as `[/]action//` and `[/]output//`, mirroring the de-facto convention so a bucket stays interoperable with other `GOCACHEPROG` implementations. ## Shared Cache Trust Model diff --git a/cmd/plaid-cache/commands.go b/cmd/plaid-cache/commands.go index 5c5b890..1298d5f 100644 --- a/cmd/plaid-cache/commands.go +++ b/cmd/plaid-cache/commands.go @@ -539,7 +539,7 @@ func (a *app) printStatus(cfg *config.Config, actions, objects, diskBytes int64, return } a.outf("daemon pid %d, up %s\n", d.PID, d.Uptime) - a.printCounters(d.Metrics, cfg.RemoteEnabled()) + a.printCounters(d, cfg.RemoteEnabled()) a.printLifetime(life, lifeSince) } @@ -573,7 +573,7 @@ func (a *app) printStatusFrom(endpoint string, r *daemon.StatusResponse) { a.outf("remote disabled\n") } a.outf("daemon pid %d, up %s\n", r.PID, r.Uptime) - a.printCounters(r.Metrics, r.RemoteEnabled) + a.printCounters(r, r.RemoteEnabled) a.printLifetime(r.Lifetime, r.LifetimeSince) } @@ -630,7 +630,13 @@ func (a *app) printAge(oldest, newest string) { // would otherwise have to work out from three separate counters. Repairs are // called out because a nonzero count means bodies went missing under the index, // which is worth noticing rather than burying. -func (a *app) printCounters(m cache.MetricsSnapshot, remoteEnabled bool) { +// +// It takes the whole report rather than the counters alone because the upload +// backlog is not one of them: it is a level, and it belongs beside the upload +// counters because it is what those counters cannot say — how close the next +// burst is to being dropped. +func (a *app) printCounters(d *daemon.StatusResponse, remoteEnabled bool) { + m := d.Metrics lookups := m.GetLocalHit + m.GetRemoteHit + m.GetMiss if lookups > 0 { a.outf("hit rate %.1f%% of %d lookups\n", @@ -648,6 +654,9 @@ func (a *app) printCounters(m cache.MetricsSnapshot, remoteEnabled bool) { if remoteEnabled { a.outf("uploads %d ok, %d failed, %d dropped, %d skipped\n", m.UploadOK, m.UploadFail, m.UploadDrop, m.UploadSkip) + if d.UploadQueueCapacity > 0 { + a.outf("upload q %d of %d queued\n", d.UploadQueueDepth, d.UploadQueueCapacity) + } } } diff --git a/internal/cache/cache.go b/internal/cache/cache.go index f35e361..15e9142 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -123,7 +123,7 @@ func New(p Params) *Cache { rem: rem, logf: logf, } - c.uploads = newUploader(p.Config.UploadConcurrency, logf, &c.metrics) + c.uploads = newUploader(p.Config.UploadConcurrency, p.Config.UploadQueueDepth, p.Config.UploadBlockTimeout, logf, &c.metrics) return c } @@ -346,6 +346,15 @@ func (c *Cache) record(a ids.ActionID, o ids.OutputID, path string, size, diskBy // Metrics returns a snapshot of the counters. func (c *Cache) Metrics() MetricsSnapshot { return c.metrics.Snapshot() } +// UploadQueue reports how many uploads are waiting and how many may wait. +// +// It is not part of Metrics because it is not the same kind of number. Those +// are counters that only rise and are persisted across processes; this is a +// level that rises and falls and means nothing once the process holding the +// queue is gone. Its use is to be watched while it is still climbing — the drop +// counter can only be read after the entries are already lost. +func (c *Cache) UploadQueue() (depth, capacity int) { return c.uploads.queue() } + // Evict runs one eviction pass, removing orphaned bodies as the index // releases them. func (c *Cache) Evict(ctx context.Context) (index.EvictResult, error) { @@ -473,6 +482,12 @@ type uploadJob struct { // of concurrent connections; an unbounded queue would let it accumulate // unbounded memory and delay process exit. When the queue is full the job is // dropped and counted, which is the correct trade for a best-effort tier. +// +// What that trade costs is paid somewhere else and much later: the entry is +// simply not in the shared tier, so a reader on another machine takes a clean +// miss and redoes the work, with nothing at the moment of the loss to connect +// the two. Hence queue and depth below, which say how close to full the queue is +// before any of it is lost, and the drop log, which says that it happened. type uploader struct { jobs chan uploadJob wg sync.WaitGroup @@ -480,6 +495,23 @@ type uploader struct { metrics *Metrics once sync.Once + // blockFor is how long submit waits for room before dropping. Zero, the + // default, never waits. + blockFor time.Duration + + // quit is closed at the start of close, before the lock a waiting submit + // holds is taken, so a bounded wait cannot hold up exit for its whole + // timeout. It is separate from jobs because closing jobs is what a waiting + // send must not observe. + quit chan struct{} + + // droppedSinceLog counts drops not yet reported, and loggedAt is when the + // last report went out, as Unix nanoseconds. Together they rate-limit the + // warning: a saturation episode is thousands of drops a second, and a line + // each would bury the log it is trying to make legible. + droppedSinceLog atomic.Int64 + loggedAt atomic.Int64 + // mu guards jobs against a send racing close. A send on a closed channel // panics even inside a select with a default, because the send case is // always ready once the channel is closed, so default cannot make submit @@ -489,18 +521,32 @@ type uploader struct { closed bool } -// queueDepthPerWorker sizes the backlog relative to the pool. +// queueDepthPerWorker sizes the backlog relative to the pool when nothing else +// says otherwise. Configured by PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH. const queueDepthPerWorker = 64 +// dropLogInterval is the shortest gap between two drop warnings. Each one +// carries the count since the last, so a longer gap loses no information about +// how much was lost, only about exactly when. +const dropLogInterval = 30 * time.Second + // newUploader starts the worker pool. -func newUploader(workers int, logf Logf, m *Metrics) *uploader { +// +// depth is per worker and blockFor may be zero, which is the default and means +// a full queue drops immediately rather than waiting. +func newUploader(workers, depth int, blockFor time.Duration, logf Logf, m *Metrics) *uploader { if workers < 1 { workers = 1 } + if depth < 1 { + depth = queueDepthPerWorker + } u := &uploader{ - jobs: make(chan uploadJob, workers*queueDepthPerWorker), - logf: logf, - metrics: m, + jobs: make(chan uploadJob, workers*depth), + logf: logf, + metrics: m, + blockFor: blockFor, + quit: make(chan struct{}), } u.wg.Add(workers) for range workers { @@ -509,21 +555,81 @@ func newUploader(workers int, logf Logf, m *Metrics) *uploader { return u } +// queue reports how much of the backlog is in use, which is the number that +// moves before anything is lost rather than after. +// +// Both halves are needed to read it: a depth of 400 is idle on one pool and +// about to start dropping on another, and the capacity is derived from the +// worker count, so a reader has no way to work it out for themselves. +func (u *uploader) queue() (depth, capacity int) { + return len(u.jobs), cap(u.jobs) +} + // submit queues a job, dropping it if the queue is full rather than blocking // the build that produced it. +// +// A configured blockFor buys a bounded wait for room first. It is off by +// default: a put that waits is a build that waits, and the whole promise of this +// tier is that it cannot do that. func (u *uploader) submit(j uploadJob) { u.mu.RLock() defer u.mu.RUnlock() if u.closed { // Nothing will drain it, so count it as dropped rather than panicking. + // Counted but not logged: a submission arriving after shutdown is not + // the queue overflowing, and saying so would send a reader looking for + // load that was never there. u.metrics.UploadDrop.Add(1) return } select { case u.jobs <- j: + return default: - u.metrics.UploadDrop.Add(1) } + if u.blockFor <= 0 { + u.dropped() + return + } + t := time.NewTimer(u.blockFor) + defer t.Stop() + select { + case u.jobs <- j: + case <-t.C: + u.dropped() + case <-u.quit: + // close is waiting on the read lock this send holds. What is already + // queued will still be drained; one more job is not worth delaying the + // process's exit by the rest of a timeout an operator chose for load, + // not for shutdown. + u.dropped() + } +} + +// dropped counts a lost upload and says so in the log, at most once per +// dropLogInterval. +// +// The count since the previous report is the point of the line. A rate is what +// distinguishes a queue that overflowed on one burst from one that has been +// shedding the shared tier's contents for an hour, and the counter alone cannot +// tell those apart without someone already watching it. +func (u *uploader) dropped() { + u.metrics.UploadDrop.Add(1) + n := u.droppedSinceLog.Add(1) + + now := time.Now().UnixNano() + last := u.loggedAt.Load() + // The first drop reports immediately: last is zero, which is further in the + // past than any interval. Losing the CAS means another goroutine is + // reporting this instant, and its line will carry this drop too. + if now-last < int64(dropLogInterval) || !u.loggedAt.CompareAndSwap(last, now) { + return + } + u.droppedSinceLog.Add(-n) + _, capacity := u.queue() + u.logf("upload queue full: dropped %d uploads since the last report; those entries are missing from the "+ + "shared tier (the queue holds %d — raise PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH or PLAID_GOCACHE_UPLOAD_CONCURRENCY)", + n, capacity) } // uploadTimeout bounds a single transfer so one stuck connection cannot hold @@ -574,6 +680,10 @@ func (u *uploader) run(j uploadJob) { // close stops accepting work and waits for the pool to drain. func (u *uploader) close() { u.once.Do(func() { + // Released before the lock is taken, since a submit part-way through a + // bounded wait holds that lock and would otherwise keep exit waiting for + // the remainder of its timeout. + close(u.quit) u.mu.Lock() u.closed = true close(u.jobs) diff --git a/internal/cache/fake_test.go b/internal/cache/fake_test.go index 0fc7922..e6d6e90 100644 --- a/internal/cache/fake_test.go +++ b/internal/cache/fake_test.go @@ -10,6 +10,7 @@ import ( "errors" "fmt" "io" + "strings" "sync" "testing" "time" @@ -197,6 +198,41 @@ type testCache struct { blobs *blob.Store rem *fakeRemote cache *Cache + logs *logRecorder +} + +// logRecorder captures what the cache logged, for the diagnostics that exist to +// be read rather than counted. Lines still reach t.Logf, which is the other +// reason they are there. +// +// mu guards lines because the uploader logs from its worker pool and from +// whichever goroutine submitted a job it had to drop. +type logRecorder struct { + t *testing.T + mu sync.Mutex + lines []string +} + +// logf records one line and forwards it to the test log. +func (r *logRecorder) logf(format string, args ...any) { + line := fmt.Sprintf(format, args...) + r.mu.Lock() + r.lines = append(r.lines, line) + r.mu.Unlock() + r.t.Log(line) +} + +// matching returns the recorded lines containing sub. +func (r *logRecorder) matching(sub string) []string { + r.mu.Lock() + defer r.mu.Unlock() + var out []string + for _, l := range r.lines { + if strings.Contains(l, sub) { + out = append(out, l) + } + } + return out } // option mutates the config before the tiers are opened. @@ -212,6 +248,17 @@ func withMinUploadSize(n int64) option { return func(c *config.Config) { c.MinUploadSize = n } } +// withUploadQueueDepth sets the per-worker upload backlog. +func withUploadQueueDepth(n int) option { + return func(c *config.Config) { c.UploadQueueDepth = n } +} + +// withUploadBlockTimeout makes a full upload queue wait for room rather than +// drop immediately. +func withUploadBlockTimeout(d time.Duration) option { + return func(c *config.Config) { c.UploadBlockTimeout = d } +} + // withCompactAfter sets the pruned-entry debt that triggers a compaction. func withCompactAfter(n int64) option { return func(c *config.Config) { c.CompactAfterPruned = n } @@ -251,12 +298,13 @@ func newTestCache(t *testing.T, opts ...option) *testCache { } rem := newFakeRemote() - c := New(Params{Config: cfg, Index: idx, Blobs: blobs, Remote: rem, Logf: t.Logf}) + logs := &logRecorder{t: t} + c := New(Params{Config: cfg, Index: idx, Blobs: blobs, Remote: rem, Logf: logs.logf}) // Registered after the index cleanup so it runs first: the uploader must // drain while the index is still open. t.Cleanup(func() { _ = c.Close() }) - return &testCache{cfg: cfg, idx: idx, blobs: blobs, rem: rem, cache: c} + return &testCache{cfg: cfg, idx: idx, blobs: blobs, rem: rem, cache: c, logs: logs} } // put is a terse Cache.Put for tests, returning the on-disk path. diff --git a/internal/cache/upload_test.go b/internal/cache/upload_test.go index 4cc2233..43230a9 100644 --- a/internal/cache/upload_test.go +++ b/internal/cache/upload_test.go @@ -7,6 +7,7 @@ import ( "bytes" "fmt" "slices" + "strings" "sync" "testing" "time" @@ -95,6 +96,185 @@ func TestUploadWritesObjectBeforeAction(t *testing.T) { } } +// pinPool holds every upload inside PutObject until the returned function is +// called, so a test can fill the queue without racing the workers. +// +// The release is registered as a cleanup ahead of whatever the caller does with +// it, because newTestCache's own Close cleanup runs later and would otherwise +// wedge on a worker nobody let go. +func pinPool(t *testing.T, tc *testCache) func() { + t.Helper() + release := make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + t.Cleanup(unblock) + tc.rem.blockPutObject = release + return unblock +} + +// TestUploadQueueReportsSaturationBeforeDrops pins the gauge that arrives in +// time to be acted on: with the pool pinned, the backlog is readable while it is +// filling and still whole, not only once the queue has overflowed and entries +// the shared tier will never hold are already gone. +func TestUploadQueueReportsSaturationBeforeDrops(t *testing.T) { + const depth = 8 + tc := newTestCache(t, withRemote(), withUploadQueueDepth(depth)) + unblock := pinPool(t, tc) + + if got, capacity := tc.cache.UploadQueue(); got != 0 || capacity != depth { + t.Fatalf("UploadQueue = %d of %d on an idle cache, want 0 of %d", got, capacity, depth) + } + + // Fewer submissions than the queue holds: one is in the pinned worker's + // hands and the rest are waiting, so nothing can have been dropped yet. + body := bytes.Repeat([]byte("q"), 32) + for i := range depth - 1 { + tc.put(t, mkActionN(i), mkOutputN(i), body) + } + got, capacity := tc.cache.UploadQueue() + if capacity != depth { + t.Fatalf("UploadQueue capacity = %d, want the configured %d", capacity, depth) + } + if got == 0 || got > capacity { + t.Fatalf("UploadQueue = %d of %d with %d uploads pinned, want a depth in (0, %d]", + got, capacity, depth-1, capacity) + } + if m := tc.cache.Metrics(); m.UploadDrop != 0 { + t.Fatalf("UploadDrop = %d below capacity, want 0: the depth must be readable before anything is lost", m.UploadDrop) + } + + // Past capacity the queue can only report itself full, which is why the + // depth above is the number worth watching. + for i := depth; i < 4*depth; i++ { + tc.put(t, mkActionN(i), mkOutputN(i), body) + } + if got, capacity := tc.cache.UploadQueue(); got != capacity { + t.Fatalf("UploadQueue = %d of %d after oversubscribing, want a full queue", got, capacity) + } + if m := tc.cache.Metrics(); m.UploadDrop == 0 { + t.Fatalf("UploadDrop = 0 after oversubscribing a queue of %d, want drops", depth) + } + unblock() +} + +// TestUploadDropWarningIsRateLimited pins that a saturation episode says so in +// the log exactly once per interval, however many uploads it loses. +// +// Both halves matter. Silence is what made this failure mode invisible — the +// entries are simply absent, and the reader that misses them is on another +// machine much later. A line per drop would be no better: thousands of them a +// second bury the log they were meant to make legible, so the count since the +// last report travels in the line instead. +func TestUploadDropWarningIsRateLimited(t *testing.T) { + const depth = 2 + tc := newTestCache(t, withRemote(), withUploadQueueDepth(depth)) + unblock := pinPool(t, tc) + + body := bytes.Repeat([]byte("d"), 32) + const submitted = 40 + for i := range submitted { + tc.put(t, mkActionN(i), mkOutputN(i), body) + } + + m := tc.cache.Metrics() + if m.UploadDrop < 2 { + t.Fatalf("UploadDrop = %d, want several: the test cannot pin rate limiting without repeated drops", m.UploadDrop) + } + lines := tc.logs.matching("upload queue full") + if len(lines) != 1 { + t.Fatalf("logged %d drop warnings for %d drops within one %v interval, want 1: %v", + len(lines), m.UploadDrop, dropLogInterval, lines) + } + // The first drop reports immediately and carries the one it is reporting; + // a warning that named no count would leave a reader unable to tell a + // single overflow from a sustained one. + if !strings.Contains(lines[0], "dropped 1 upload") { + t.Fatalf("drop warning %q does not report how many were dropped", lines[0]) + } + unblock() +} + +// TestUploadBlockTimeoutWaitsForRoom pins the opt-in reversal of the drop trade: +// with a timeout configured, a put whose queue is full waits for room instead of +// losing the entry. +func TestUploadBlockTimeoutWaitsForRoom(t *testing.T) { + tc := newTestCache(t, withRemote(), withUploadQueueDepth(1), withUploadBlockTimeout(blockDeadline)) + unblock := pinPool(t, tc) + + // One job pins the worker and one fills the queue, so the third has nowhere + // to go and would be dropped without a timeout to wait out. + body := bytes.Repeat([]byte("b"), 32) + tc.put(t, mkActionN(0), mkOutputN(0), body) + tc.put(t, mkActionN(1), mkOutputN(1), body) + + done := make(chan error, 1) + go func() { + _, err := tc.cache.Put(t.Context(), mkActionN(2), mkOutputN(2), bytes.NewReader(body), int64(len(body))) + done <- err + }() + + // Nothing can drain while the pool is pinned, so a Put that returns here + // returned by dropping — which is what the timeout exists to prevent. + select { + case err := <-done: + unblock() + t.Fatalf("Put returned (err %v) with the queue full and the pool pinned, want it waiting for room", err) + case <-time.After(50 * time.Millisecond): + } + + unblock() + select { + case err := <-done: + if err != nil { + t.Fatalf("Put: %v", err) + } + case <-time.After(blockDeadline): + t.Fatalf("Put blocked for %v after the pool was released, want it to land", blockDeadline) + } + + if err := tc.cache.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + if m := tc.cache.Metrics(); m.UploadDrop != 0 || m.UploadOK != 3 { + t.Fatalf("UploadOK = %d, UploadDrop = %d, want 3 and 0: a waited-for slot must not lose the entry", + m.UploadOK, m.UploadDrop) + } +} + +// TestUploadBlockTimeoutExpiresRatherThanStalling pins the bound on that wait. A +// wait with no end would turn a saturated queue into a stalled build, which is +// the failure the drop was chosen to avoid in the first place. +func TestUploadBlockTimeoutExpiresRatherThanStalling(t *testing.T) { + tc := newTestCache(t, withRemote(), withUploadQueueDepth(1), withUploadBlockTimeout(20*time.Millisecond)) + unblock := pinPool(t, tc) + + body := bytes.Repeat([]byte("e"), 32) + done := make(chan error, 1) + go func() { + for i := range 4 { + if _, err := tc.cache.Put(t.Context(), mkActionN(i), mkOutputN(i), bytes.NewReader(body), int64(len(body))); err != nil { + done <- fmt.Errorf("Put(%d): %w", i, err) + return + } + } + done <- nil + }() + + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(blockDeadline): + unblock() + t.Fatalf("Put blocked for %v against a 20ms upload timeout, want the job dropped", blockDeadline) + } + if m := tc.cache.Metrics(); m.UploadDrop == 0 { + t.Fatal("UploadDrop = 0 with the pool pinned past the block timeout, want the waits to have expired") + } + unblock() +} + // TestUploadQueueDropsRatherThanBlocking pins the bounded-queue trade: with the // pool pinned on a stuck transfer, submissions past the queue depth are dropped // and counted, and no Put ever blocks on the backlog. diff --git a/internal/config/config.go b/internal/config/config.go index 2fdae2b..22d750a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -114,6 +114,29 @@ type Config struct { // UploadConcurrency bounds in-flight remote uploads. UploadConcurrency int + // UploadQueueDepth is how many uploads may wait per worker before further + // ones are dropped. It is per worker rather than absolute so that the + // backlog scales with the pool draining it, the way it always has. + // + // It is a setting rather than a constant because the right value is a + // property of the machine and not of the tool: the queue holds a job per + // entry, so raising it buys tolerance of a burst with memory, and a builder + // that produces entries faster than one link can ship them wants a different + // number from a laptop. Dropping is still what happens when it fills; this + // only decides how much of a burst passes without one. + UploadQueueDepth int + + // UploadBlockTimeout is how long a put waits for room in the upload queue + // before its upload is dropped. Zero, the default, never waits. + // + // Waiting is opt-in because the default is the trade a best-effort tier + // should make: a full queue costs the shared tier an entry, and no build + // stalls on a cache. Setting this reverses that for the callers who would + // rather pay latency than lose the entry — a nightly warmer filling a bucket + // for everyone else, where a dropped upload is the whole point of the run — + // and it is bounded so the reversal cannot become an unbounded stall. + UploadBlockTimeout time.Duration + // IdleTimeout is how long the daemon runs with no connected clients // before exiting. IdleTimeout time.Duration @@ -193,6 +216,7 @@ const ( defaultTTL = 168 * time.Hour defaultTouchGranularity = time.Hour defaultMinUploadSize = 0 + defaultUploadQueueDepth = 64 defaultIdleTimeout = 30 * time.Minute defaultEvictInterval = time.Minute defaultCompactAfter = 1000 @@ -258,6 +282,15 @@ func Load() (*Config, error) { if c.UploadConcurrency < 1 { return nil, fmt.Errorf("Load: PLAID_GOCACHE_UPLOAD_CONCURRENCY: got %d, want >= 1", c.UploadConcurrency) } + if c.UploadQueueDepth, err = envInt(src, "PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH", defaultUploadQueueDepth); err != nil { + return nil, fmt.Errorf("Load: %w", err) + } + if c.UploadQueueDepth < 1 { + return nil, fmt.Errorf("Load: PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH: got %d, want >= 1", c.UploadQueueDepth) + } + if c.UploadBlockTimeout, err = envDuration(src, "PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT", 0); err != nil { + return nil, fmt.Errorf("Load: %w", err) + } if c.BazelMonitoring, err = envBool(src, "PLAID_GOCACHE_BAZEL_MONITORING"); err != nil { return nil, fmt.Errorf("Load: %w", err) } @@ -304,6 +337,8 @@ var settingNames = map[string]bool{ "PLAID_GOCACHE_S3_ENDPOINT_URL": true, "PLAID_GOCACHE_MIN_UPLOAD_SIZE": true, "PLAID_GOCACHE_UPLOAD_CONCURRENCY": true, + "PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH": true, + "PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT": true, "PLAID_GOCACHE_TOUCH_GRANULARITY": true, "PLAID_GOCACHE_IDLE_TIMEOUT": true, "PLAID_GOCACHE_EVICT_INTERVAL": true, diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 1620269..6fc4d18 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -110,6 +110,15 @@ func TestLoadDefaults(t *testing.T) { if c.DisableBazelVerify { t.Fatalf("DisableBazelVerify = true by default, want uploads verified") } + if c.UploadQueueDepth != defaultUploadQueueDepth { + t.Fatalf("UploadQueueDepth = %d, want %d", c.UploadQueueDepth, defaultUploadQueueDepth) + } + // Waiting for room in the upload queue is opt-in. A default that blocked + // would turn a full queue from a lost cache entry into a stalled build, + // which is the one thing this tool promises not to do. + if c.UploadBlockTimeout != 0 { + t.Fatalf("UploadBlockTimeout = %v by default, want 0 (drop rather than wait)", c.UploadBlockTimeout) + } } // TestLoadDirChain pins the PLAID_GOCACHE_DIR > XDG_CACHE_HOME precedence. @@ -148,6 +157,10 @@ func TestLoadRejectsBadValues(t *testing.T) { {"PLAID_GOCACHE_MIN_UPLOAD_SIZE", "small"}, {"PLAID_GOCACHE_UPLOAD_CONCURRENCY", "many"}, {"PLAID_GOCACHE_UPLOAD_CONCURRENCY", "0"}, + {"PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH", "deep"}, + {"PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH", "0"}, + {"PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT", "a while"}, + {"PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT", "-1s"}, {"PLAID_GOCACHE_IDLE_TIMEOUT", "forever"}, {"PLAID_GOCACHE_EVICT_INTERVAL", "often"}, {"PLAID_GOCACHE_DISABLE_EVICTION", "true"}, @@ -267,6 +280,8 @@ func clearEnv(t *testing.T) { "PLAID_GOCACHE_S3_ENDPOINT_URL", "PLAID_GOCACHE_MIN_UPLOAD_SIZE", "PLAID_GOCACHE_UPLOAD_CONCURRENCY", + "PLAID_GOCACHE_UPLOAD_QUEUE_DEPTH", + "PLAID_GOCACHE_UPLOAD_BLOCK_TIMEOUT", "PLAID_GOCACHE_IDLE_TIMEOUT", "PLAID_GOCACHE_EVICT_INTERVAL", "PLAID_GOCACHE_DISABLE_EVICTION", diff --git a/internal/daemon/metrics.go b/internal/daemon/metrics.go index 5835282..c2b5a40 100644 --- a/internal/daemon/metrics.go +++ b/internal/daemon/metrics.go @@ -80,6 +80,20 @@ func renderMetrics(r StatusResponse) []byte { sample{value: r.NewestAgeSeconds}) } + // The upload backlog, which is the one number here that leads its failure + // rather than recording it. A queue at its capacity is dropping uploads, and + // the shared tier quietly stops receiving what this machine builds; the + // counter that says so is uploads_total{result="dropped"}, and by the time it + // moves the entries are gone. These two are a process level rather than a + // persisted total for the same reason they are worth having: what is queued + // belongs to the daemon holding it and means nothing after it exits. + writeFamily(&b, "upload_queue_depth", "gauge", + "Uploads to the shared tier waiting on this daemon's worker pool.", + sample{value: float64(r.UploadQueueDepth)}) + writeFamily(&b, "upload_queue_capacity", "gauge", + "Uploads that may wait before further ones are dropped. Depth at capacity means entries are being lost.", + sample{value: float64(r.UploadQueueCapacity)}) + writeFamily(&b, "uptime_seconds", "gauge", "Time since this daemon started.", sample{value: r.UptimeSeconds}) writeFamily(&b, "remote_tier_enabled", "gauge", diff --git a/internal/daemon/metrics_test.go b/internal/daemon/metrics_test.go index a11a538..35785eb 100644 --- a/internal/daemon/metrics_test.go +++ b/internal/daemon/metrics_test.go @@ -156,6 +156,10 @@ func TestMetricsExposition(t *testing.T) { NewestAge: "1s", NewestAgeSeconds: 1, LifetimeSince: 1_700_000_000_000_000_000, + + UploadQueueDepth: 12, + UploadQueueCapacity: 512, + Lifetime: cache.MetricsSnapshot{ GetLocalHit: 205, GetRemoteHit: 3, GetMiss: 139, GetRepair: 2, Put: 275, UploadOK: 205, UploadFail: 1, UploadDrop: 2, UploadSkip: 3, @@ -170,6 +174,10 @@ func TestMetricsExposition(t *testing.T) { "plaid_cache_oldest_entry_age_seconds", "plaid_cache_newest_entry_age_seconds", "plaid_cache_remote_tier_enabled", "plaid_cache_activity_start_time_seconds", "plaid_cache_build_info", + // A backlog rises and falls, so a counter here would make every + // dashboard built on it nonsense — and this is the one family meant to + // be watched climbing, before the drops it predicts have happened. + "plaid_cache_upload_queue_depth", "plaid_cache_upload_queue_capacity", } { e.wantType(t, f, "gauge") } @@ -204,6 +212,8 @@ func TestMetricsExposition(t *testing.T) { e.wantSample(t, `plaid_cache_uploads_total{result="dropped"}`, 2) e.wantSample(t, `plaid_cache_uploads_total{result="skipped"}`, 3) e.wantSample(t, "plaid_cache_compactions_total", 7) + e.wantSample(t, "plaid_cache_upload_queue_depth", 12) + e.wantSample(t, "plaid_cache_upload_queue_capacity", 512) // Every family this daemon exposes carries the one prefix, and nothing // carries a label whose values are not a fixed, short list — a label per diff --git a/internal/daemon/protocol.go b/internal/daemon/protocol.go index 9a377cd..bcaa1fa 100644 --- a/internal/daemon/protocol.go +++ b/internal/daemon/protocol.go @@ -157,6 +157,16 @@ type StatusResponse struct { Metrics cache.MetricsSnapshot `json:"metrics"` Err string `json:"err,omitempty"` + // UploadQueueDepth is how many uploads to the shared tier are waiting on + // this daemon's pool right now, and UploadQueueCapacity how many may. + // + // They are here rather than in Metrics because they are levels rather than + // counters, and because they are what says saturation is coming: the drop + // counter beside them only moves once entries have already been lost, and + // says nothing about a queue that is one burst away from losing them. + UploadQueueDepth int `json:"upload_queue_depth"` + UploadQueueCapacity int `json:"upload_queue_capacity"` + // The durations above again, in seconds. // // Each pair is two encodings of one measurement taken at one moment, not two diff --git a/internal/daemon/server.go b/internal/daemon/server.go index 9e781c4..6ca10f3 100644 --- a/internal/daemon/server.go +++ b/internal/daemon/server.go @@ -604,6 +604,7 @@ func (s *Server) status() StatusResponse { UptimeSeconds: uptime.Seconds(), Metrics: s.cache.Metrics(), } + r.UploadQueueDepth, r.UploadQueueCapacity = s.cache.UploadQueue() // The lifetime figures include what this process has counted but not yet // flushed, so the two sets cannot disagree about activity that just // happened — a lifetime total smaller than the session's would be a diff --git a/internal/daemon/stats_op_test.go b/internal/daemon/stats_op_test.go index df491ce..68ec8fb 100644 --- a/internal/daemon/stats_op_test.go +++ b/internal/daemon/stats_op_test.go @@ -173,4 +173,13 @@ func TestStatusReportsLifetimeAlongsideTheSession(t *testing.T) { if st.LifetimeSince == 0 { t.Fatal("no start time for the lifetime figures") } + // The upload backlog travels with the report rather than only with the + // counters, because it is the number that says saturation is coming rather + // than that it has already cost something. + if st.UploadQueueCapacity <= 0 { + t.Fatalf("upload queue capacity = %d, want the daemon's own bound", st.UploadQueueCapacity) + } + if st.UploadQueueDepth != 0 { + t.Fatalf("upload queue depth = %d on an idle daemon, want 0", st.UploadQueueDepth) + } }