gocached: skip the cleanup intent for blobs already in the main dir

gocache/gocached/gocached.go

@@ -1180,6 +1180,7 @@ type Server struct {
 		PutQueueBlockedSpoolBytes expvar.Int `type:"counter" name:"put_queue_blocked_spool_bytes" help:"PUT requests that waited on the spooled lane's byte cap; raise the spool capacity if the disk has room, or look at the movers if pending never drains"`
 		PutQueueBlockedSpoolCount expvar.Int `type:"counter" name:"put_queue_blocked_spool_count" help:"PUT requests that waited on the spooled lane's count cap; many small blobs are piling up faster than the movers copy them"`
 		PutQueueCopyErrs          expvar.Int `type:"counter" name:"put_queue_copy_errs" help:"failed attempts to copy a spooled blob into the main blob directory; retried before the PUT is dropped"`
+		PutQueueCopySkips         expvar.Int `type:"counter" name:"put_queue_copy_skips" help:"spooled blobs already present in the main blob directory with the expected size, so neither copied nor given a cleanup intent; a subset of gocached_put_queue_copy_duration_seconds_count"`
 		PutQueueDropped           expvar.Int `type:"counter" name:"put_queue_dropped" help:"pending PUTs abandoned after repeated copy or flush failures; the client saw success but the object was lost"`
 		PutQueueFlushes           expvar.Int `type:"counter" name:"put_queue_flushes" help:"metadata batch transactions committed by the put-queue flusher"`
 		PutQueueFlushedItems      expvar.Int `type:"counter" name:"put_queue_flushed_items" help:"pending PUTs whose metadata was committed by the put-queue flusher"`

gocache/gocached/putqueue.go

@@ -492,16 +492,13 @@ func (q *putQueue) moveToFlush(ctx context.Context, p *pendingPut) {
 	q.drop(p)
 }
 
-// copyTimed performs one attempt at p's intent write and blob copy into the
-// main blob directory and records the attempt's wall time.
+// copyTimed performs one attempt at installing p's blob in the main blob
+// directory and records the attempt's wall time.
 func (q *putQueue) copyTimed(p *pendingPut) error {
 	q.activeCopies.Add(1)
 	defer q.activeCopies.Add(-1)
 	start := time.Now()
-	err := q.writeCleanupIntent(p)
-	if err == nil {
-		err = q.copyToMain(p)
-	}
+	err := q.installInMain(p)
 	if q.srv.putCopyDuration != nil {
 		result := "ok"
 		if err != nil {
@@ -512,6 +509,36 @@ func (q *putQueue) copyTimed(p *pendingPut) error {
 	return err
 }
 
+// installInMain makes p's blob present in the main blob directory.
+//
+// A blob already there with the expected size is left alone, since blobs
+// are content-addressed, and it gets no cleanup intent: this PUT creates no
+// file that the sweeper might have to reap. An intent wouldn't protect the
+// existing blob anyway. The sweeper keeps any blob that SQLite accounts
+// for or that a pending PUT references, which covers p until its metadata
+// commits. In production most spooled blobs are already present, and on a
+// network filesystem the intent's create and later unlink are the costly
+// part: they are mutations of the one cleanup directory, and the kernel
+// serializes those.
+//
+// Otherwise it writes p's cleanup intent and then copies the blob.
+func (q *putQueue) installInMain(p *pendingPut) error {
+	if fi, err := os.Stat(q.mainPath(p)); err == nil && fi.Size() == p.storedSize {
+		q.srv.m.PutQueueCopySkips.Add(1)
+		return nil
+	}
+	if err := q.writeCleanupIntent(p); err != nil {
+		return err
+	}
+	return q.copyToMain(p)
+}
+
+// mainPath returns the path of p's blob in the main blob directory.
+func (q *putQueue) mainPath(p *pendingPut) string {
+	name := p.blobName()
+	return filepath.Join(q.srv.dir, name[:2], name)
+}
+
 // writeCleanupIntent creates p's cleanup intent record: a durable note that
 // p's blob is about to exist in the main blob directory without SQLite
 // accounting, and should be deleted at the encoded time if it's still
@@ -633,18 +660,14 @@ func (q *putQueue) pendingSHA(sha string) bool {
 	return false
 }
 
-// copyToMain installs p's spooled blob into the main blob directory, leaving
+// copyToMain copies p's spooled blob into the main blob directory, leaving
 // the spool file in place: it remains the bytes served for GETs of the
 // pending entry, and finishCommitted may later rename it into the hot tier.
-// A blob already present with the expected size is left alone; blobs are
-// content-addressed, so it holds the same bytes.
+// Callers use installInMain, which skips the copy if the blob is already
+// present.
 func (q *putQueue) copyToMain(p *pendingPut) error {
 	srv := q.srv
-	name := p.blobName()
-	target := filepath.Join(srv.dir, name[:2], name)
-	if fi, err := os.Stat(target); err == nil && fi.Size() == p.storedSize {
-		return nil
-	}
+	target := q.mainPath(p)
 
 	src, err := os.Open(p.queueFile)
 	if err != nil {
@@ -872,11 +895,7 @@ empty:
 	toFlush := batch[:0]
 	for _, p := range batch {
 		if p.queueFile != "" {
-			err := q.writeCleanupIntent(p)
-			if err == nil {
-				err = q.copyToMain(p)
-			}
-			if err != nil {
+			if err := q.installInMain(p); err != nil {
 				srv.logf("put-queue: copying blob %v to main dir: %v", p.sha256hex, err)
 				srv.m.PutQueueCopyErrs.Add(1)
 				q.drop(p)

gocache/gocached/putqueue_test.go

@@ -384,6 +384,68 @@ func TestPutQueueDrain(t *testing.T) {
 	st.wantMetric(&st.srv.m.PutQueueFlushes, 0)
 }
 
+// TestPutQueueSkipIntentWhenPresent verifies that a spooled blob already in
+// the main blob directory is neither copied nor given a cleanup intent, and
+// still commits, while a new blob gets its intent as usual.
+func TestPutQueueSkipIntentWhenPresent(t *testing.T) {
+	st := newServerTester(t)
+	q := st.srv.putq
+	content := bytes.Repeat([]byte("present"), 1000)
+
+	p1 := makePending(t, q, 0, "ee01", content)
+	q.enqueue(p1)
+	st.drain()
+	st.wantMetric(&st.srv.m.PutQueueCopySkips, 0)
+	fi1, err := os.Stat(q.mainPath(p1))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// A different action with the same output: its blob is already there.
+	p2 := makePending(t, q, 0, "ee02", content)
+	q.enqueue(p2)
+	if err := q.installInMain(p2); err != nil {
+		t.Fatal(err)
+	}
+	if p2.intentPath != "" {
+		t.Errorf("intent %s written for an already-present blob", p2.intentPath)
+	}
+	if ents, err := os.ReadDir(q.cleanupDir); err != nil || len(ents) != 0 {
+		t.Errorf("cleanup dir has %d entries, err=%v; want none", len(ents), err)
+	}
+	st.wantMetric(&st.srv.m.PutQueueCopySkips, 1)
+	if fi2, err := os.Stat(q.mainPath(p2)); err != nil || !os.SameFile(fi1, fi2) {
+		t.Errorf("present blob was replaced (err=%v)", err)
+	}
+
+	// A new blob still gets an intent before its copy.
+	p3 := makePending(t, q, 0, "ee03", bytes.Repeat([]byte("new"), 1000))
+	q.enqueue(p3)
+	if err := q.installInMain(p3); err != nil {
+		t.Fatal(err)
+	}
+	if p3.intentPath == "" {
+		t.Fatal("no intent written for a new blob")
+	}
+	if _, err := os.Stat(p3.intentPath); err != nil {
+		t.Errorf("intent for new blob: %v", err)
+	}
+	st.wantMetric(&st.srv.m.PutQueueCopySkips, 0)
+
+	// Both commit.
+	st.drain()
+	var n int
+	if err := st.srv.db.QueryRow("SELECT COUNT(*) FROM Actions").Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 3 {
+		t.Errorf("Actions rows = %d, want 3", n)
+	}
+	if _, err := os.Stat(q.mainPath(p3)); err != nil {
+		t.Errorf("new blob not installed: %v", err)
+	}
+}
+
 func TestPutQueueDrainFlushDup(t *testing.T) {
 	st := newServerTester(t)
 	q := st.srv.putq