chore: log ingest queue, merge and store telemetry

Author: mennatnaga

internal/db/merge.go

@@ -14,7 +14,9 @@ import (
 	"container/list"
 	"context"
 	"fmt"
+	"log/slog"
 	"sort"
+	"strings"
 	"sync"
 
 	"github.com/ipfs/go-cid"
@@ -76,6 +78,12 @@ func (db *DB) Merge(ctx context.Context, evt event.Merge) error {
 // several, so a transaction's cost grows with the square of the events in it.
 const mergeChunkSize = 8
 
+// Phases a merge chunk can fail in, reported on the retry-exhaustion log line.
+const (
+	phaseRead   = "read"
+	phaseCommit = "commit"
+)
+
 type mergeEntry struct {
 	evt event.Merge
 	col *collection
@@ -197,11 +205,25 @@ func (db *DB) txnAttempts() int {
 	return 1
 }
 
+// namedDocs renders the documents a transaction touched, as collection/docID. Badger
+// reports conflicts without naming the contended key, so this is the only lead available
+// for working out which documents contend with each other.
+func namedDocs(entries []mergeEntry) string {
+	docIDs := make([]string, len(entries))
+	for i, e := range entries {
+		docIDs[i] = e.col.Name() + "/" + e.evt.DocID
+	}
+	return strings.Join(docIDs, ",")
+}
+
 // mergeChunk merges every event of the chunk inside one transaction, retrying the
 // whole chunk on transaction conflict. Isolating a failing event is the caller's job.
 func (db *DB) mergeChunk(ctx context.Context, entries []mergeEntry) error {
-	// Held so that exhausting the retry budget can report the conflict that caused it.
+	// Held so that exhausting the retry budget can report the conflict that caused it,
+	// along with where the last one was raised and which event raised it.
 	var conflictErr error
+	var blamed mergeEntry
+	var phase string
 	for i := 0; i < db.txnAttempts(); i++ {
 		txn, err := db.NewTxn(false)
 		if err != nil {
@@ -212,6 +234,7 @@ func (db *DB) mergeChunk(ctx context.Context, entries []mergeEntry) error {
 		var mergeErr error
 		for _, e := range entries {
 			if mergeErr = db.mergeInTxn(txnCtx, e.col, e.evt); mergeErr != nil {
+				blamed, phase = e, phaseRead
 				break
 			}
 		}
@@ -229,6 +252,9 @@ func (db *DB) mergeChunk(ctx context.Context, entries []mergeEntry) error {
 			txn.Discard()
 			if errors.Is(err, corekv.ErrTxnConflict) {
 				conflictErr = err
+				// A commit conflict belongs to the whole transaction and cannot be
+				// attributed to one event.
+				blamed, phase = mergeEntry{}, phaseCommit
 				continue
 			}
 			return err
@@ -237,7 +263,20 @@ func (db *DB) mergeChunk(ctx context.Context, entries []mergeEntry) error {
 		return nil
 	}
 
-	// Nothing was committed, so callers must not treat the events as merged.
+	// Nothing was committed, so callers must not treat the events as merged. This is the
+	// one place a conflict becomes data loss, so it is reported even though the retries
+	// leading to it are not.
+	fields := []slog.Attr{
+		corelog.Int("attempts", db.MaxTxnRetries()),
+		corelog.String("phase", phase),
+		corelog.String("docIDs", namedDocs(entries)),
+	}
+	// Only a read conflict has a single event to blame; at commit it would just repeat
+	// the chunk, and these lines are already long.
+	if phase == phaseRead {
+		fields = append(fields, corelog.String("blamed", namedDocs([]mergeEntry{blamed})))
+	}
+	log.InfoContext(ctx, "merge chunk exhausted its retries", fields...)
 	return client.NewErrMaxTxnRetries(conflictErr)
 }
 

internal/db/p2p/p2p.go

@@ -82,6 +82,11 @@ const (
 	// msgQueueFallbackMaxBytes is the byte budget used when the process has no memory
 	// limit set, where a fraction of it would be meaningless.
 	msgQueueFallbackMaxBytes = 1 << 30
+
+	// statsInterval is how often queue depth and merge counters are reported. Rates are
+	// reported per interval rather than per event, which keeps the ingest path quiet
+	// under load where per-event logging would dominate the output.
+	statsInterval = 30 * time.Second
 )
 
 // queuedMessage is a decoded request paired with its wire size, so the queue's byte
@@ -206,6 +211,19 @@ type P2P struct {
 	msgQueueBytes atomic.Int64
 	// msgQueueMaxBytes caps msgQueueBytes. Zero or less disables the byte bound.
 	msgQueueMaxBytes int64
+
+	// stopStats ends the stats reporter. The context outlives Close, so the reporter
+	// needs a signal of its own.
+	stopStats chan struct{}
+
+	// Counters reported by reportStats and reset on each report, so each line carries
+	// the rate for that interval rather than a running total.
+	statDroppedBudget atomic.Int64
+	statDroppedFull   atomic.Int64
+	statMergedDocs    atomic.Int64
+	statDroppedDocs   atomic.Int64
+	statBatches       atomic.Int64
+	statBatchFailures atomic.Int64
 }
 
 // pushLogCommProcessor implements CommProcessor for push log functionality
@@ -263,12 +281,14 @@ func New(
 		topicPeerCounts:      make(map[string]int),
 		msgQueue:             make(chan queuedMessage, msgQueueSize),
 		msgQueueMaxBytes:     queueByteBudget(),
+		stopStats:            make(chan struct{}),
 	}
 
 	for i := 0; i < dagSyncWorkers; i++ {
 		p.msgWorkers.Add(1)
 		go p.processMessageWorker()
 	}
+	go p.reportStats()
 
 	p.replicatorProtocol = protocol.NewCommChannel(host, "rep", &pushLogCommProcessor{p2p: &p})
 	p.batcher = newPubsubBatcher(host.ID(), func(topic string, data []byte) error {
@@ -661,6 +681,7 @@ func (p *P2P) docIDsForBlockCID(
 func (p *P2P) pubSubMessageHandler(from string, topic string, msg []byte) ([]byte, error) {
 	size := int64(len(msg))
 	if !p.claimQueueBytes(size) {
+		p.statDroppedBudget.Add(1)
 		log.Info("pubsub message queue over byte budget, dropping message",
 			corelog.Any("topic", topic),
 			corelog.Int64("bytes", size),
@@ -681,6 +702,7 @@ func (p *P2P) pubSubMessageHandler(from string, topic string, msg []byte) ([]byt
 		p.releaseQueueBytes(size)
 	default:
 		p.releaseQueueBytes(size)
+		p.statDroppedFull.Add(1)
 		log.Info("pubsub message queue full, dropping message", corelog.Any("topic", topic))
 	}
 	return nil, nil
@@ -707,6 +729,35 @@ func (p *P2P) releaseQueueBytes(size int64) {
 	p.msgQueueBytes.Add(-size)
 }
 
+// reportStats periodically logs queue occupancy and merge outcomes. Queue depth and the
+// split between merged and dropped documents are otherwise only visible from a heap
+// profile, which is too invasive to take routinely.
+func (p *P2P) reportStats() {
+	ticker := time.NewTicker(statsInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-p.ctx.Done():
+			return
+		case <-p.stopStats:
+			return
+		case <-ticker.C:
+			log.Info("p2p stats",
+				corelog.Int("queueDepth", len(p.msgQueue)),
+				corelog.Int("queueSlots", msgQueueSize),
+				corelog.Int64("queueBytes", p.msgQueueBytes.Load()),
+				corelog.Int64("queueBudget", p.msgQueueMaxBytes),
+				corelog.Int64("droppedOverBudget", p.statDroppedBudget.Swap(0)),
+				corelog.Int64("droppedQueueFull", p.statDroppedFull.Swap(0)),
+				corelog.Int64("batches", p.statBatches.Swap(0)),
+				corelog.Int64("batchFailures", p.statBatchFailures.Swap(0)),
+				corelog.Int64("docsMerged", p.statMergedDocs.Swap(0)),
+				corelog.Int64("docsDropped", p.statDroppedDocs.Swap(0)),
+			)
+		}
+	}
+}
+
 // processMessageWorker is a long-lived goroutine that drains p.msgQueue and
 // calls processPushlogRequest for each message. dagSyncWorkers of these run
 // concurrently, bounding the goroutine count regardless of inbound message rate.
@@ -772,7 +823,16 @@ func (p *P2P) processPushlogRequest(
 			merges[i] = r.merge
 		}
 		merged, err := p.db.MergeBatchWithTxn(ctx, merges)
+		p.statBatches.Add(1)
+		for _, ok := range merged {
+			if ok {
+				p.statMergedDocs.Add(1)
+			} else {
+				p.statDroppedDocs.Add(1)
+			}
+		}
 		if err != nil {
+			p.statBatchFailures.Add(1)
 			log.ErrorE("Failed to merge documents in batch", err,
 				corelog.String("PeerID", req.SenderID),
 				corelog.Int("Documents", len(merges)))
@@ -877,8 +937,10 @@ func (p *P2P) processPushlogRequest(
 			CollectionID: req.CollectionID,
 		}
 		if err = p.db.Merge(ctx, mergeEvt); err != nil {
+			p.statDroppedDocs.Add(1)
 			return err
 		}
+		p.statMergedDocs.Add(1)
 
 		// Notify bus subscribers and the network of peers that we have a new document available.
 		updateEvt := event.Update{
@@ -1144,6 +1206,7 @@ func (pq *processQueue) close() {
 // and waits for all worker goroutines to exit.
 // It should be called once when the P2P subsystem is shutting down.
 func (p *P2P) Close() {
+	close(p.stopStats)
 	close(p.msgQueue)
 	done := make(chan struct{})
 	go func() {

internal/db/p2p/p2p_test.go

@@ -77,6 +77,8 @@ func TestPubSubMessageHandler_OverBudgetDropsBeforeDecode(t *testing.T) {
 	assert.Nil(t, resp)
 	assert.Empty(t, p.msgQueue)
 	assert.Equal(t, int64(0), p.msgQueueBytes.Load())
+	assert.Equal(t, int64(1), p.statDroppedBudget.Load())
+	assert.Equal(t, int64(0), p.statDroppedFull.Load())
 }
 
 func TestPubSubMessageHandler_ByteBudgetReleasedAfterProcessing(t *testing.T) {
@@ -142,6 +144,32 @@ func TestProcessMessageWorker_ReleasesByteBudget(t *testing.T) {
 		"the worker must return the budget the handler reserved")
 }
 
+// A full queue and an exhausted budget are separate drop reasons, and the stats line is
+// only useful if a drop is attributed to the one that caused it.
+func TestPubSubMessageHandler_QueueFullCountedSeparately(t *testing.T) {
+	msg, err := cbor.Marshal(protocol.PushLogRequest{DocID: "docID"})
+	assert.NoError(t, err)
+
+	// Budget large enough to stay out of the way, so only the slot count can reject.
+	p := &P2P{
+		ctx:              context.Background(),
+		host:             &SimpleMockHost{},
+		msgQueue:         make(chan queuedMessage, 1),
+		msgQueueMaxBytes: 1 << 20,
+	}
+
+	_, err = p.pubSubMessageHandler("sender", "topic", msg)
+	assert.NoError(t, err)
+
+	_, err = p.pubSubMessageHandler("sender", "topic", msg)
+	assert.NoError(t, err)
+
+	assert.Equal(t, int64(1), p.statDroppedFull.Load())
+	assert.Equal(t, int64(0), p.statDroppedBudget.Load())
+	// The dropped message must not keep holding its reservation.
+	assert.Equal(t, int64(len(msg)), p.msgQueueBytes.Load())
+}
+
 func TestQueueByteBudget_DerivesFromMemoryLimit(t *testing.T) {
 	original := debug.SetMemoryLimit(-1)
 	t.Cleanup(func() { debug.SetMemoryLimit(original) })

node/store_badger.go

@@ -145,6 +145,9 @@ func (s *badgerStore) runValueLogGC() {
 // reclaim, an error occurs, or the store starts closing. Each successful call
 // rewrites one file; ErrNoRewrite means no file was eligible and ends the loop.
 // Any other error is logged, since it means GC could not make progress.
+//
+// The on-disk size is reported on every pass, reclaimed or not, so whether the store is
+// bounded can be read from the log without shell access to the volume.
 func (s *badgerStore) reclaimValueLog() {
 	start := time.Now()
 	reclaimed := 0
@@ -162,9 +165,10 @@ func (s *badgerStore) reclaimValueLog() {
 		}
 		reclaimed++
 	}
-	if reclaimed > 0 {
-		log.Info("Reclaimed badger value log files",
-			corelog.Int("files", reclaimed),
-			corelog.Duration("duration", time.Since(start)))
-	}
+	lsm, vlog := s.db.Size()
+	log.Info("Badger value log GC",
+		corelog.Int("files", reclaimed),
+		corelog.Int64("lsmBytes", lsm),
+		corelog.Int64("vlogBytes", vlog),
+		corelog.Duration("duration", time.Since(start)))
 }