Say when the upload queue is filling, not only after it has lost something

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 `[<prefix>/]action/<xx>/<hex-action-id>` and `[<prefix>/]output/<xx>/<hex-output-id>`, mirroring the de-facto convention so a bucket stays interoperable with other `GOCACHEPROG` implementations.
 
 ## Shared Cache Trust Model

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)
+		}
 	}
 }
 

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,13 +482,36 @@ 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
 	logf    Logf
 	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)

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.