feat: harden launch readiness with versioning, metrics, and resilience

10 files changed, 980 insertions, 42 deletions

diff --git a/internal/api/observability.go b/internal/api/observability.go
new file mode 100644
index 0000000..ebd7711
--- /dev/null
+++ b/internal/api/observability.go
@@ -0,0 +1,293 @@
+package api
+
+import (
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/Fuwn/plutia/internal/config"
+	"github.com/Fuwn/plutia/internal/ingest"
+	"github.com/Fuwn/plutia/internal/storage"
+	"github.com/Fuwn/plutia/internal/types"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+type BuildInfo struct {
+	Version   string `json:"version"`
+	Commit    string `json:"commit"`
+	BuildDate string `json:"build_date"`
+	GoVersion string `json:"go_version"`
+}
+
+type serverMetrics struct {
+	registry           *prometheus.Registry
+	checkpointDuration prometheus.Histogram
+	checkpointSequence prometheus.Gauge
+}
+
+func newServerMetrics(cfg config.Config, store storage.Store, ingestor *ingest.Service) *serverMetrics {
+	reg := prometheus.NewRegistry()
+	var diskMu sync.Mutex
+	var diskCached int64
+	var diskCachedAt time.Time
+	m := &serverMetrics{
+		registry: reg,
+		checkpointDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Name:    "checkpoint_duration_seconds",
+			Help:    "Time spent generating and signing checkpoints.",
+			Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 20, 60},
+		}),
+		checkpointSequence: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "checkpoint_sequence",
+			Help: "Latest checkpoint sequence generated by this mirror.",
+		}),
+	}
+	reg.MustRegister(m.checkpointDuration, m.checkpointSequence)
+	reg.MustRegister(prometheus.NewCounterFunc(
+		prometheus.CounterOpts{
+			Name: "ingest_ops_total",
+			Help: "Total operations persisted by the mirror.",
+		},
+		func() float64 {
+			seq, err := store.GetGlobalSeq()
+			if err != nil {
+				return 0
+			}
+			return float64(seq)
+		},
+	))
+	reg.MustRegister(prometheus.NewGaugeFunc(
+		prometheus.GaugeOpts{
+			Name: "ingest_ops_per_second",
+			Help: "Ingestion operations per second (process-average).",
+		},
+		func() float64 {
+			if ingestor == nil {
+				return 0
+			}
+			return ingestor.Stats().IngestOpsPerSec
+		},
+	))
+	reg.MustRegister(prometheus.NewGaugeFunc(
+		prometheus.GaugeOpts{
+			Name: "ingest_lag_ops",
+			Help: "Difference between latest observed upstream sequence and committed global sequence.",
+		},
+		func() float64 {
+			if ingestor == nil {
+				return 0
+			}
+			return float64(ingestor.Stats().LagOps)
+		},
+	))
+	reg.MustRegister(prometheus.NewCounterFunc(
+		prometheus.CounterOpts{
+			Name: "verify_failures_total",
+			Help: "Total signature/link verification failures seen during ingestion.",
+		},
+		func() float64 {
+			if ingestor == nil {
+				return 0
+			}
+			return float64(ingestor.Stats().VerifyFailures)
+		},
+	))
+	reg.MustRegister(prometheus.NewGaugeFunc(
+		prometheus.GaugeOpts{
+			Name: "disk_bytes_total",
+			Help: "Total bytes used by the configured data directory.",
+		},
+		func() float64 {
+			diskMu.Lock()
+			defer diskMu.Unlock()
+			if diskCachedAt.IsZero() || time.Since(diskCachedAt) > 5*time.Second {
+				size, err := dirSize(cfg.DataDir)
+				if err == nil {
+					diskCached = size
+					diskCachedAt = time.Now()
+				}
+			}
+			return float64(diskCached)
+		},
+	))
+	reg.MustRegister(prometheus.NewGaugeFunc(
+		prometheus.GaugeOpts{
+			Name: "did_count",
+			Help: "Number of DIDs currently materialized in state storage.",
+		},
+		func() float64 {
+			if ingestor != nil {
+				return float64(ingestor.Stats().DIDCount)
+			}
+			count := uint64(0)
+			_ = store.ForEachState(func(_ types.StateV1) error {
+				count++
+				return nil
+			})
+			return float64(count)
+		},
+	))
+	return m
+}
+
+func (m *serverMetrics) Handler() http.Handler {
+	return promhttp.HandlerFor(m.registry, promhttp.HandlerOpts{
+		EnableOpenMetrics: true,
+	})
+}
+
+func (m *serverMetrics) ObserveCheckpoint(duration time.Duration, sequence uint64) {
+	m.checkpointDuration.Observe(duration.Seconds())
+	m.checkpointSequence.Set(float64(sequence))
+}
+
+type serverOption func(*Server)
+
+func WithBuildInfo(info BuildInfo) serverOption {
+	return func(s *Server) {
+		s.build = info
+	}
+}
+
+type limiterClass int
+
+const (
+	limiterResolve limiterClass = iota
+	limiterProof
+)
+
+type tokenBucket struct {
+	tokens   float64
+	last     time.Time
+	lastSeen time.Time
+}
+
+type endpointPolicy struct {
+	rps   float64
+	burst float64
+}
+
+type ipRateLimiter struct {
+	mu        sync.Mutex
+	buckets   map[string]*tokenBucket
+	resolve   endpointPolicy
+	proof     endpointPolicy
+	lastSweep time.Time
+}
+
+func newIPRateLimiter(cfg config.RateLimit) *ipRateLimiter {
+	def := config.Default().RateLimit
+	if cfg.ResolveRPS <= 0 {
+		cfg.ResolveRPS = def.ResolveRPS
+	}
+	if cfg.ResolveBurst <= 0 {
+		cfg.ResolveBurst = def.ResolveBurst
+	}
+	if cfg.ProofRPS <= 0 {
+		cfg.ProofRPS = def.ProofRPS
+	}
+	if cfg.ProofBurst <= 0 {
+		cfg.ProofBurst = def.ProofBurst
+	}
+	return &ipRateLimiter{
+		buckets: map[string]*tokenBucket{},
+		resolve: endpointPolicy{
+			rps:   cfg.ResolveRPS,
+			burst: float64(cfg.ResolveBurst),
+		},
+		proof: endpointPolicy{
+			rps:   cfg.ProofRPS,
+			burst: float64(cfg.ProofBurst),
+		},
+		lastSweep: time.Now(),
+	}
+}
+
+func (l *ipRateLimiter) Allow(ip string, class limiterClass) bool {
+	now := time.Now()
+	l.mu.Lock()
+	defer l.mu.Unlock()
+
+	if now.Sub(l.lastSweep) > 2*time.Minute {
+		for key, bucket := range l.buckets {
+			if now.Sub(bucket.lastSeen) > 15*time.Minute {
+				delete(l.buckets, key)
+			}
+		}
+		l.lastSweep = now
+	}
+
+	policy := l.resolve
+	routeKey := "resolve"
+	if class == limiterProof {
+		policy = l.proof
+		routeKey = "proof"
+	}
+	key := routeKey + "|" + ip
+	b, ok := l.buckets[key]
+	if !ok {
+		l.buckets[key] = &tokenBucket{
+			tokens:   policy.burst - 1,
+			last:     now,
+			lastSeen: now,
+		}
+		return true
+	}
+	elapsed := now.Sub(b.last).Seconds()
+	if elapsed > 0 {
+		b.tokens += elapsed * policy.rps
+		if b.tokens > policy.burst {
+			b.tokens = policy.burst
+		}
+	}
+	b.last = now
+	b.lastSeen = now
+	if b.tokens < 1 {
+		return false
+	}
+	b.tokens--
+	return true
+}
+
+func clientIP(r *http.Request) string {
+	if forwarded := strings.TrimSpace(r.Header.Get("X-Forwarded-For")); forwarded != "" {
+		parts := strings.Split(forwarded, ",")
+		if len(parts) > 0 {
+			if ip := strings.TrimSpace(parts[0]); ip != "" {
+				return ip
+			}
+		}
+	}
+	if realIP := strings.TrimSpace(r.Header.Get("X-Real-IP")); realIP != "" {
+		return realIP
+	}
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err == nil && host != "" {
+		return host
+	}
+	return r.RemoteAddr
+}
+
+func dirSize(path string) (int64, error) {
+	var total int64
+	err := filepath.WalkDir(path, func(_ string, d os.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			return nil
+		}
+		info, err := d.Info()
+		if err != nil {
+			return err
+		}
+		total += info.Size()
+		return nil
+	})
+	return total, err
+}
diff --git a/internal/api/server.go b/internal/api/server.go
index 2159029..f1ffc7c 100644
--- a/internal/api/server.go
+++ b/internal/api/server.go
@@ -1,12 +1,15 @@
 package api
 
 import (
+	"context"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/Fuwn/plutia/internal/checkpoint"
 	"github.com/Fuwn/plutia/internal/config"
@@ -22,20 +25,46 @@ type Server struct {
 	store       storage.Store
 	ingestor    *ingest.Service
 	checkpoints *checkpoint.Manager
+	build       BuildInfo
+	limiter     *ipRateLimiter
+	metrics     *serverMetrics
 }
 
-func NewServer(cfg config.Config, store storage.Store, ingestor *ingest.Service, checkpoints *checkpoint.Manager) *Server {
-	return &Server{cfg: cfg, store: store, ingestor: ingestor, checkpoints: checkpoints}
+func NewServer(cfg config.Config, store storage.Store, ingestor *ingest.Service, checkpoints *checkpoint.Manager, opts ...serverOption) *Server {
+	s := &Server{
+		cfg:         cfg,
+		store:       store,
+		ingestor:    ingestor,
+		checkpoints: checkpoints,
+		build: BuildInfo{
+			Version:   "dev",
+			Commit:    "unknown",
+			BuildDate: "unknown",
+			GoVersion: "unknown",
+		},
+		limiter: newIPRateLimiter(cfg.RateLimit),
+	}
+	for _, opt := range opts {
+		opt(s)
+	}
+	s.metrics = newServerMetrics(cfg, store, ingestor)
+	if cp, ok, err := store.GetLatestCheckpoint(); err == nil && ok {
+		s.metrics.checkpointSequence.Set(float64(cp.Sequence))
+	}
+	if ingestor != nil {
+		ingestor.SetMetricsSink(s.metrics)
+	}
+	return s
 }
 
 func (s *Server) Handler() http.Handler {
 	mux := http.NewServeMux()
-	mux.HandleFunc("/health", s.handleHealth)
-	mux.HandleFunc("/metrics", s.handleMetrics)
-	mux.HandleFunc("/status", s.handleStatus)
-	mux.HandleFunc("/checkpoints/latest", s.handleLatestCheckpoint)
-	mux.HandleFunc("/checkpoints/", s.handleCheckpointBySequence)
-	mux.HandleFunc("/did/", s.handleDID)
+	mux.Handle("/health", s.withTimeout(http.HandlerFunc(s.handleHealth)))
+	mux.Handle("/metrics", s.metrics.Handler())
+	mux.Handle("/status", s.withTimeout(http.HandlerFunc(s.handleStatus)))
+	mux.Handle("/checkpoints/latest", s.withTimeout(http.HandlerFunc(s.handleLatestCheckpoint)))
+	mux.Handle("/checkpoints/", s.withTimeout(http.HandlerFunc(s.handleCheckpointBySequence)))
+	mux.Handle("/did/", s.withTimeout(http.HandlerFunc(s.handleDID)))
 	return mux
 }
 
@@ -43,15 +72,6 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, map[string]any{"status": "ok"})
 }
 
-func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) {
-	st := s.ingestor.Stats()
-	seq, _ := s.store.GetGlobalSeq()
-	w.Header().Set("Content-Type", "text/plain; version=0.0.4")
-	_, _ = fmt.Fprintf(w, "plutia_ingested_ops %d\n", st.IngestedOps)
-	_, _ = fmt.Fprintf(w, "plutia_ingest_errors %d\n", st.Errors)
-	_, _ = fmt.Fprintf(w, "plutia_last_seq %d\n", seq)
-}
-
 func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
 	seq, err := s.store.GetGlobalSeq()
 	if err != nil {
@@ -63,15 +83,22 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
 		writeErr(w, http.StatusInternalServerError, err)
 		return
 	}
+	stats := ingest.Stats{}
+	if s.ingestor != nil {
+		stats = s.ingestor.Stats()
+	}
 	payload := map[string]any{
 		"mode":          s.cfg.Mode,
 		"verify_policy": s.cfg.VerifyPolicy,
 		"global_seq":    seq,
-		"stats":         s.ingestor.Stats(),
-		"corrupted":     s.ingestor.IsCorrupted(),
+		"stats":         stats,
+		"build":         s.build,
 	}
-	if err := s.ingestor.CorruptionError(); err != nil {
-		payload["corruption_error"] = err.Error()
+	if s.ingestor != nil {
+		payload["corrupted"] = s.ingestor.IsCorrupted()
+		if err := s.ingestor.CorruptionError(); err != nil {
+			payload["corruption_error"] = err.Error()
+		}
 	}
 	if ok {
 		payload["latest_checkpoint"] = cp
@@ -123,9 +150,17 @@ func (s *Server) handleDID(w http.ResponseWriter, r *http.Request) {
 	}
 	if strings.HasSuffix(path, "/proof") {
 		did := strings.TrimSuffix(path, "/proof")
+		if !s.allowRequest(r, limiterProof) {
+			writeErr(w, http.StatusTooManyRequests, fmt.Errorf("proof rate limit exceeded"))
+			return
+		}
 		s.handleDIDProof(w, r, did)
 		return
 	}
+	if !s.allowRequest(r, limiterResolve) {
+		writeErr(w, http.StatusTooManyRequests, fmt.Errorf("resolve rate limit exceeded"))
+		return
+	}
 	s.handleDIDResolve(w, r, path)
 }
 
@@ -159,6 +194,10 @@ func (s *Server) handleDIDResolve(w http.ResponseWriter, r *http.Request, did st
 }
 
 func (s *Server) handleDIDProof(w http.ResponseWriter, r *http.Request, did string) {
+	if s.ingestor == nil {
+		writeErr(w, http.StatusServiceUnavailable, fmt.Errorf("ingestor unavailable"))
+		return
+	}
 	if err := s.ingestor.CorruptionError(); err != nil {
 		writeErr(w, http.StatusServiceUnavailable, err)
 		return
@@ -172,11 +211,19 @@ func (s *Server) handleDIDProof(w http.ResponseWriter, r *http.Request, did stri
 
 	tipHash, seqs, err := s.ingestor.RecomputeTipAtOrBefore(r.Context(), did, cp.Sequence)
 	if err != nil {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			writeErr(w, http.StatusGatewayTimeout, err)
+			return
+		}
 		writeErr(w, http.StatusNotFound, err)
 		return
 	}
-	siblings, leafHash, found, err := s.checkpoints.BuildDIDProofAtCheckpoint(did, tipHash, cp.Sequence)
+	siblings, leafHash, found, err := s.checkpoints.BuildDIDProofAtCheckpoint(r.Context(), did, tipHash, cp.Sequence)
 	if err != nil {
+		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			writeErr(w, http.StatusGatewayTimeout, err)
+			return
+		}
 		writeErr(w, http.StatusInternalServerError, err)
 		return
 	}
@@ -228,6 +275,25 @@ func (s *Server) handleDIDProof(w http.ResponseWriter, r *http.Request, did stri
 	})
 }
 
+func (s *Server) withTimeout(next http.Handler) http.Handler {
+	timeout := s.cfg.RequestTimeout
+	if timeout <= 0 {
+		timeout = 10 * time.Second
+	}
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx, cancel := context.WithTimeout(r.Context(), timeout)
+		defer cancel()
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}
+
+func (s *Server) allowRequest(r *http.Request, class limiterClass) bool {
+	if s.limiter == nil {
+		return true
+	}
+	return s.limiter.Allow(clientIP(r), class)
+}
+
 func (s *Server) selectCheckpointForProof(r *http.Request) (types.CheckpointV1, func() error, error) {
 	checkpointParam := strings.TrimSpace(r.URL.Query().Get("checkpoint"))
 	if checkpointParam == "" {
diff --git a/internal/api/server_hardening_test.go b/internal/api/server_hardening_test.go
new file mode 100644
index 0000000..bb9f24c
--- /dev/null
+++ b/internal/api/server_hardening_test.go
@@ -0,0 +1,119 @@
+package api
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/Fuwn/plutia/internal/config"
+	"github.com/Fuwn/plutia/internal/storage"
+	"github.com/Fuwn/plutia/internal/types"
+)
+
+func TestResolveRateLimitPerIP(t *testing.T) {
+	store, err := storage.OpenPebble(t.TempDir())
+	if err != nil {
+		t.Fatalf("open pebble: %v", err)
+	}
+	defer store.Close()
+	if err := store.PutState(types.StateV1{Version: 1, DID: "did:plc:alice", DIDDocument: []byte(`{"id":"did:plc:alice"}`), ChainTipHash: "tip", LatestOpSeq: 1, UpdatedAt: time.Now().UTC()}); err != nil {
+		t.Fatalf("put state: %v", err)
+	}
+
+	cfg := config.Default()
+	cfg.RequestTimeout = 10 * time.Second
+	cfg.RateLimit.ResolveRPS = 1
+	cfg.RateLimit.ResolveBurst = 1
+	cfg.RateLimit.ProofRPS = 1
+	cfg.RateLimit.ProofBurst = 1
+	h := NewServer(cfg, store, nil, nil).Handler()
+
+	req1 := httptest.NewRequest(http.MethodGet, "/did/did:plc:alice", nil)
+	req1.RemoteAddr = "203.0.113.7:12345"
+	rr1 := httptest.NewRecorder()
+	h.ServeHTTP(rr1, req1)
+	if rr1.Code != http.StatusOK {
+		t.Fatalf("first request status: got %d want %d", rr1.Code, http.StatusOK)
+	}
+
+	req2 := httptest.NewRequest(http.MethodGet, "/did/did:plc:alice", nil)
+	req2.RemoteAddr = "203.0.113.7:12345"
+	rr2 := httptest.NewRecorder()
+	h.ServeHTTP(rr2, req2)
+	if rr2.Code != http.StatusTooManyRequests {
+		t.Fatalf("second request status: got %d want %d", rr2.Code, http.StatusTooManyRequests)
+	}
+}
+
+func TestStatusIncludesBuildInfo(t *testing.T) {
+	store, err := storage.OpenPebble(t.TempDir())
+	if err != nil {
+		t.Fatalf("open pebble: %v", err)
+	}
+	defer store.Close()
+
+	h := NewServer(config.Default(), store, nil, nil, WithBuildInfo(BuildInfo{
+		Version:   "v0.1.0",
+		Commit:    "abc123",
+		BuildDate: "2026-02-26T00:00:00Z",
+		GoVersion: "go1.test",
+	})).Handler()
+
+	req := httptest.NewRequest(http.MethodGet, "/status", nil)
+	rr := httptest.NewRecorder()
+	h.ServeHTTP(rr, req)
+	if rr.Code != http.StatusOK {
+		t.Fatalf("status code: got %d want %d", rr.Code, http.StatusOK)
+	}
+
+	var payload map[string]any
+	if err := json.Unmarshal(rr.Body.Bytes(), &payload); err != nil {
+		t.Fatalf("decode status: %v", err)
+	}
+	build, ok := payload["build"].(map[string]any)
+	if !ok {
+		t.Fatalf("missing build section: %v", payload)
+	}
+	if got := build["version"]; got != "v0.1.0" {
+		t.Fatalf("unexpected build version: %v", got)
+	}
+	if got := build["commit"]; got != "abc123" {
+		t.Fatalf("unexpected build commit: %v", got)
+	}
+}
+
+func TestMetricsExposeRequiredSeries(t *testing.T) {
+	store, err := storage.OpenPebble(t.TempDir())
+	if err != nil {
+		t.Fatalf("open pebble: %v", err)
+	}
+	defer store.Close()
+
+	h := NewServer(config.Default(), store, nil, nil).Handler()
+	req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
+	rr := httptest.NewRecorder()
+	h.ServeHTTP(rr, req)
+	if rr.Code != http.StatusOK {
+		t.Fatalf("metrics status: got %d want %d", rr.Code, http.StatusOK)
+	}
+	body, _ := io.ReadAll(rr.Body)
+	text := string(body)
+	for _, metric := range []string{
+		"ingest_ops_total",
+		"ingest_ops_per_second",
+		"ingest_lag_ops",
+		"verify_failures_total",
+		"checkpoint_duration_seconds",
+		"checkpoint_sequence",
+		"disk_bytes_total",
+		"did_count",
+	} {
+		if !strings.Contains(text, metric) {
+			t.Fatalf("metrics output missing %q", metric)
+		}
+	}
+}
diff --git a/internal/checkpoint/checkpoint.go b/internal/checkpoint/checkpoint.go
index 840fac3..c3ed0da 100644
--- a/internal/checkpoint/checkpoint.go
+++ b/internal/checkpoint/checkpoint.go
@@ -2,6 +2,7 @@ package checkpoint
 
 import (
 	"bufio"
+	"context"
 	"crypto/ed25519"
 	"crypto/sha256"
 	"encoding/base64"
@@ -107,7 +108,7 @@ func (m *Manager) signAndPersist(sequence uint64, didRoot string, blockHashes []
 	return unsigned, nil
 }
 
-func (m *Manager) BuildDIDProofAtCheckpoint(did, chainTipHash string, checkpointSeq uint64) ([]merkle.Sibling, string, bool, error) {
+func (m *Manager) BuildDIDProofAtCheckpoint(ctx context.Context, did, chainTipHash string, checkpointSeq uint64) ([]merkle.Sibling, string, bool, error) {
 	snapshot, err := m.LoadStateSnapshot(checkpointSeq)
 	if err != nil {
 		return nil, "", false, err
@@ -116,6 +117,9 @@ func (m *Manager) BuildDIDProofAtCheckpoint(did, chainTipHash string, checkpoint
 	index := -1
 	leafHashHex := ""
 	for i, s := range snapshot.Leaves {
+		if err := ctx.Err(); err != nil {
+			return nil, "", false, err
+		}
 		h := merkle.HashLeaf([]byte(s.DID + s.ChainTipHash))
 		leaves[i] = h
 		if s.DID == did && s.ChainTipHash == chainTipHash {
diff --git a/internal/checkpoint/checkpoint_test.go b/internal/checkpoint/checkpoint_test.go
index 9ca9b2a..8149129 100644
--- a/internal/checkpoint/checkpoint_test.go
+++ b/internal/checkpoint/checkpoint_test.go
@@ -1,9 +1,11 @@
 package checkpoint
 
 import (
+	"context"
 	"crypto/ed25519"
 	"crypto/rand"
 	"encoding/base64"
+	"encoding/json"
 	"os"
 	"path/filepath"
 	"testing"
@@ -103,3 +105,52 @@ func TestCheckpointRootStability(t *testing.T) {
 		t.Fatalf("block root changed for identical block set: %s vs %s", cp1.BlockMerkleRoot, cp2.BlockMerkleRoot)
 	}
 }
+
+func TestBuildDIDProofAtCheckpointHonorsContextCancellation(t *testing.T) {
+	tmp := t.TempDir()
+	store, err := storage.OpenPebble(tmp)
+	if err != nil {
+		t.Fatalf("open pebble: %v", err)
+	}
+	defer store.Close()
+
+	_, priv, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+	keyPath := filepath.Join(tmp, "mirror.key")
+	if err := os.WriteFile(keyPath, []byte(base64.RawURLEncoding.EncodeToString(priv)), 0o600); err != nil {
+		t.Fatalf("write key: %v", err)
+	}
+	checkpointsDir := filepath.Join(tmp, "checkpoints")
+	if err := os.MkdirAll(checkpointsDir, 0o755); err != nil {
+		t.Fatalf("mkdir checkpoints: %v", err)
+	}
+	leaves := make([]types.DIDLeaf, 500)
+	for i := range leaves {
+		leaves[i] = types.DIDLeaf{
+			DID:          "did:plc:test-" + string(rune('a'+(i%26))),
+			ChainTipHash: "tip",
+		}
+	}
+	snapshot := types.CheckpointStateSnapshotV1{
+		Version:   1,
+		Sequence:  10,
+		CreatedAt: "2026-02-26T00:00:00Z",
+		Leaves:    leaves,
+	}
+	b, err := json.Marshal(snapshot)
+	if err != nil {
+		t.Fatalf("marshal snapshot: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(checkpointsDir, "00000000000000000010.state.json"), b, 0o644); err != nil {
+		t.Fatalf("write snapshot: %v", err)
+	}
+
+	mgr := NewManager(store, tmp, keyPath)
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, _, _, err := mgr.BuildDIDProofAtCheckpoint(ctx, "did:plc:test-a", "tip", 10); err == nil {
+		t.Fatalf("expected context cancellation error")
+	}
+}
diff --git a/internal/config/config.go b/internal/config/config.go
index 5e6e467..609bde5 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -34,6 +34,18 @@ type Config struct {
 	ListenAddr           string        `yaml:"listen_addr"`
 	MirrorPrivateKeyPath string        `yaml:"mirror_private_key_path"`
 	PollInterval         time.Duration `yaml:"poll_interval"`
+	RequestTimeout       time.Duration `yaml:"request_timeout"`
+	RateLimit            RateLimit     `yaml:"rate_limit"`
+	HTTPRetryMaxAttempts int           `yaml:"http_retry_max_attempts"`
+	HTTPRetryBaseDelay   time.Duration `yaml:"http_retry_base_delay"`
+	HTTPRetryMaxDelay    time.Duration `yaml:"http_retry_max_delay"`
+}
+
+type RateLimit struct {
+	ResolveRPS   float64 `yaml:"resolve_rps"`
+	ResolveBurst int     `yaml:"resolve_burst"`
+	ProofRPS     float64 `yaml:"proof_rps"`
+	ProofBurst   int     `yaml:"proof_burst"`
 }
 
 func Default() Config {
@@ -50,6 +62,16 @@ func Default() Config {
 		ListenAddr:           ":8080",
 		MirrorPrivateKeyPath: "./mirror.key",
 		PollInterval:         5 * time.Second,
+		RequestTimeout:       10 * time.Second,
+		RateLimit: RateLimit{
+			ResolveRPS:   30,
+			ResolveBurst: 60,
+			ProofRPS:     10,
+			ProofBurst:   20,
+		},
+		HTTPRetryMaxAttempts: 8,
+		HTTPRetryBaseDelay:   250 * time.Millisecond,
+		HTTPRetryMaxDelay:    10 * time.Second,
 	}
 }
 
@@ -77,6 +99,13 @@ func applyEnv(cfg *Config) {
 			*dst = v
 		}
 	}
+	setFloat64 := func(key string, dst *float64) {
+		if v := strings.TrimSpace(os.Getenv(key)); v != "" {
+			if n, err := strconv.ParseFloat(v, 64); err == nil {
+				*dst = n
+			}
+		}
+	}
 	setInt := func(key string, dst *int) {
 		if v := strings.TrimSpace(os.Getenv(key)); v != "" {
 			if n, err := strconv.Atoi(v); err == nil {
@@ -111,6 +140,14 @@ func applyEnv(cfg *Config) {
 	setString("PLUTIA_LISTEN_ADDR", &cfg.ListenAddr)
 	setString("PLUTIA_MIRROR_PRIVATE_KEY_PATH", &cfg.MirrorPrivateKeyPath)
 	setDuration("PLUTIA_POLL_INTERVAL", &cfg.PollInterval)
+	setDuration("PLUTIA_REQUEST_TIMEOUT", &cfg.RequestTimeout)
+	setFloat64("PLUTIA_RATE_LIMIT_RESOLVE_RPS", &cfg.RateLimit.ResolveRPS)
+	setInt("PLUTIA_RATE_LIMIT_RESOLVE_BURST", &cfg.RateLimit.ResolveBurst)
+	setFloat64("PLUTIA_RATE_LIMIT_PROOF_RPS", &cfg.RateLimit.ProofRPS)
+	setInt("PLUTIA_RATE_LIMIT_PROOF_BURST", &cfg.RateLimit.ProofBurst)
+	setInt("PLUTIA_HTTP_RETRY_MAX_ATTEMPTS", &cfg.HTTPRetryMaxAttempts)
+	setDuration("PLUTIA_HTTP_RETRY_BASE_DELAY", &cfg.HTTPRetryBaseDelay)
+	setDuration("PLUTIA_HTTP_RETRY_MAX_DELAY", &cfg.HTTPRetryMaxDelay)
 }
 
 func (c Config) Validate() error {
@@ -149,5 +186,32 @@ func (c Config) Validate() error {
 	if c.PollInterval <= 0 {
 		return errors.New("poll_interval must be > 0")
 	}
+	if c.RequestTimeout <= 0 {
+		return errors.New("request_timeout must be > 0")
+	}
+	if c.RateLimit.ResolveRPS <= 0 {
+		return errors.New("rate_limit.resolve_rps must be > 0")
+	}
+	if c.RateLimit.ResolveBurst <= 0 {
+		return errors.New("rate_limit.resolve_burst must be > 0")
+	}
+	if c.RateLimit.ProofRPS <= 0 {
+		return errors.New("rate_limit.proof_rps must be > 0")
+	}
+	if c.RateLimit.ProofBurst <= 0 {
+		return errors.New("rate_limit.proof_burst must be > 0")
+	}
+	if c.HTTPRetryMaxAttempts < 1 || c.HTTPRetryMaxAttempts > 32 {
+		return fmt.Errorf("http_retry_max_attempts must be between 1 and 32, got %d", c.HTTPRetryMaxAttempts)
+	}
+	if c.HTTPRetryBaseDelay <= 0 {
+		return errors.New("http_retry_base_delay must be > 0")
+	}
+	if c.HTTPRetryMaxDelay <= 0 {
+		return errors.New("http_retry_max_delay must be > 0")
+	}
+	if c.HTTPRetryBaseDelay > c.HTTPRetryMaxDelay {
+		return errors.New("http_retry_base_delay must be <= http_retry_max_delay")
+	}
 	return nil
 }
diff --git a/internal/ingest/client.go b/internal/ingest/client.go
index 0f9ad43..d25b73f 100644
--- a/internal/ingest/client.go
+++ b/internal/ingest/client.go
@@ -8,11 +8,13 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"net"
 	"net/http"
 	"net/url"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"time"
 
@@ -22,9 +24,32 @@ import (
 type Client struct {
 	source string
 	http   *http.Client
+	opts   ClientOptions
 }
 
-func NewClient(source string) *Client {
+type ClientOptions struct {
+	MaxAttempts int
+	BaseDelay   time.Duration
+	MaxDelay    time.Duration
+}
+
+func NewClient(source string, opts ...ClientOptions) *Client {
+	cfg := ClientOptions{
+		MaxAttempts: 8,
+		BaseDelay:   250 * time.Millisecond,
+		MaxDelay:    10 * time.Second,
+	}
+	if len(opts) > 0 {
+		if opts[0].MaxAttempts > 0 {
+			cfg.MaxAttempts = opts[0].MaxAttempts
+		}
+		if opts[0].BaseDelay > 0 {
+			cfg.BaseDelay = opts[0].BaseDelay
+		}
+		if opts[0].MaxDelay > 0 {
+			cfg.MaxDelay = opts[0].MaxDelay
+		}
+	}
 	transport := &http.Transport{
 		Proxy:                 http.ProxyFromEnvironment,
 		DialContext:           (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}).DialContext,
@@ -37,6 +62,7 @@ func NewClient(source string) *Client {
 	}
 	return &Client{
 		source: strings.TrimRight(source, "/"),
+		opts:   cfg,
 		http: &http.Client{
 			Timeout:   60 * time.Second,
 			Transport: transport,
@@ -52,6 +78,7 @@ func (c *Client) FetchExportLimited(ctx context.Context, after uint64, limit uin
 	if strings.HasPrefix(c.source, "file://") || strings.HasSuffix(c.source, ".ndjson") || strings.HasSuffix(c.source, ".json") {
 		return c.fetchFromFile(after, limit)
 	}
+
 	u, err := url.Parse(c.source)
 	if err != nil {
 		return nil, fmt.Errorf("parse plc source: %w", err)
@@ -65,16 +92,127 @@ func (c *Client) FetchExportLimited(ctx context.Context, after uint64, limit uin
 	if err != nil {
 		return nil, fmt.Errorf("new request: %w", err)
 	}
-	resp, err := c.http.Do(req)
+	maxAttempts := c.opts.MaxAttempts
+	if maxAttempts < 1 {
+		maxAttempts = 1
+	}
+	var lastErr error
+	for attempt := 1; attempt <= maxAttempts; attempt++ {
+		records, retryAfter, retryable, err := c.fetchExportOnce(req, limit)
+		if err == nil {
+			return records, nil
+		}
+		lastErr = err
+		if !retryable || attempt == maxAttempts || ctx.Err() != nil {
+			break
+		}
+		delay := retryAfter
+		if delay <= 0 {
+			delay = c.backoffDelay(attempt)
+		}
+		log.Printf("retrying plc export fetch attempt=%d after_seq=%d delay=%s reason=%v", attempt, after, delay, err)
+		timer := time.NewTimer(delay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return nil, ctx.Err()
+		case <-timer.C:
+		}
+	}
+	return nil, lastErr
+}
+
+type httpStatusError struct {
+	StatusCode int
+	Body       string
+}
+
+func (e httpStatusError) Error() string {
+	return fmt.Sprintf("export response %d: %s", e.StatusCode, e.Body)
+}
+
+func (c *Client) fetchExportOnce(req *http.Request, limit uint64) ([]types.ExportRecord, time.Duration, bool, error) {
+	reqClone := req.Clone(req.Context())
+	reqClone.Header.Set("Accept-Encoding", "gzip")
+	resp, err := c.http.Do(reqClone)
 	if err != nil {
-		return nil, fmt.Errorf("fetch export: %w", err)
+		return nil, 0, isTransientNetworkErr(err), fmt.Errorf("fetch export: %w", err)
 	}
 	defer resp.Body.Close()
+
 	if resp.StatusCode != http.StatusOK {
 		b, _ := io.ReadAll(io.LimitReader(resp.Body, 4*1024))
-		return nil, fmt.Errorf("export response %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
+		body := strings.TrimSpace(string(b))
+		retryDelay := parseRetryAfter(resp.Header.Get("Retry-After"))
+		err := httpStatusError{StatusCode: resp.StatusCode, Body: body}
+		return nil, retryDelay, shouldRetryStatus(resp.StatusCode), err
+	}
+	records, err := decodeExportBody(resp.Body, limit)
+	if err != nil {
+		return nil, 0, false, err
+	}
+	return records, 0, false, nil
+}
+
+func (c *Client) backoffDelay(attempt int) time.Duration {
+	if attempt < 1 {
+		attempt = 1
+	}
+	delay := c.opts.BaseDelay
+	if delay <= 0 {
+		delay = 250 * time.Millisecond
+	}
+	maxDelay := c.opts.MaxDelay
+	if maxDelay <= 0 {
+		maxDelay = 10 * time.Second
+	}
+	for i := 1; i < attempt; i++ {
+		delay *= 2
+		if delay >= maxDelay {
+			return maxDelay
+		}
+	}
+	if delay > maxDelay {
+		return maxDelay
+	}
+	return delay
+}
+
+func shouldRetryStatus(status int) bool {
+	switch status {
+	case http.StatusTooManyRequests, http.StatusInternalServerError, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
+		return true
+	default:
+		return false
+	}
+}
+
+func parseRetryAfter(v string) time.Duration {
+	v = strings.TrimSpace(v)
+	if v == "" {
+		return 0
+	}
+	if secs, err := strconv.Atoi(v); err == nil && secs > 0 {
+		return time.Duration(secs) * time.Second
+	}
+	if ts, err := http.ParseTime(v); err == nil {
+		delay := time.Until(ts)
+		if delay > 0 {
+			return delay
+		}
+	}
+	return 0
+}
+
+func isTransientNetworkErr(err error) bool {
+	if err == nil {
+		return false
+	}
+	var netErr net.Error
+	if errors.As(err, &netErr) {
+		return true
 	}
-	return decodeExportBody(resp.Body, limit)
+	return errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF)
 }
 
 func decodeExportBody(r io.Reader, limit uint64) ([]types.ExportRecord, error) {
diff --git a/internal/ingest/client_retry_test.go b/internal/ingest/client_retry_test.go
new file mode 100644
index 0000000..a23fceb
--- /dev/null
+++ b/internal/ingest/client_retry_test.go
@@ -0,0 +1,63 @@
+package ingest
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestFetchExportLimitedRetries429ThenSucceeds(t *testing.T) {
+	var attempts atomic.Int32
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		n := attempts.Add(1)
+		if n <= 2 {
+			w.Header().Set("Retry-After", "0")
+			http.Error(w, "rate limited", http.StatusTooManyRequests)
+			return
+		}
+		_, _ = fmt.Fprintln(w, `{"seq":1,"did":"did:plc:alice","cid":"cid1","operation":{"x":1}}`)
+	}))
+	defer ts.Close()
+
+	client := NewClient(ts.URL, ClientOptions{
+		MaxAttempts: 5,
+		BaseDelay:   time.Millisecond,
+		MaxDelay:    2 * time.Millisecond,
+	})
+	records, err := client.FetchExportLimited(context.Background(), 0, 0)
+	if err != nil {
+		t.Fatalf("fetch export: %v", err)
+	}
+	if len(records) != 1 {
+		t.Fatalf("record count mismatch: got %d want 1", len(records))
+	}
+	if got := attempts.Load(); got != 3 {
+		t.Fatalf("attempt count mismatch: got %d want 3", got)
+	}
+}
+
+func TestFetchExportLimitedDoesNotRetry400(t *testing.T) {
+	var attempts atomic.Int32
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		attempts.Add(1)
+		http.Error(w, "bad request", http.StatusBadRequest)
+	}))
+	defer ts.Close()
+
+	client := NewClient(ts.URL, ClientOptions{
+		MaxAttempts: 5,
+		BaseDelay:   time.Millisecond,
+		MaxDelay:    2 * time.Millisecond,
+	})
+	_, err := client.FetchExportLimited(context.Background(), 0, 0)
+	if err == nil {
+		t.Fatalf("expected error for 400 response")
+	}
+	if got := attempts.Load(); got != 1 {
+		t.Fatalf("unexpected retries on 400: got attempts=%d want 1", got)
+	}
+}
diff --git a/internal/ingest/service.go b/internal/ingest/service.go
index 8451246..817a8bd 100644
--- a/internal/ingest/service.go
+++ b/internal/ingest/service.go
@@ -24,9 +24,14 @@ import (
 )
 
 type Stats struct {
-	IngestedOps uint64 `json:"ingested_ops"`
-	Errors      uint64 `json:"errors"`
-	LastSeq     uint64 `json:"last_seq"`
+	IngestedOps     uint64  `json:"ingested_ops"`
+	Errors          uint64  `json:"errors"`
+	LastSeq         uint64  `json:"last_seq"`
+	VerifyFailures  uint64  `json:"verify_failures"`
+	LagOps          uint64  `json:"lag_ops"`
+	DIDCount        uint64  `json:"did_count"`
+	CheckpointSeq   uint64  `json:"checkpoint_sequence"`
+	IngestOpsPerSec float64 `json:"ingest_ops_per_second"`
 }
 
 type Service struct {
@@ -43,6 +48,12 @@ type Service struct {
 	runOps      uint64
 	corrupted   atomic.Bool
 	corruptErr  atomic.Value
+	startedAt   time.Time
+	metricsSink MetricsSink
+}
+
+type MetricsSink interface {
+	ObserveCheckpoint(duration time.Duration, sequence uint64)
 }
 
 func NewService(cfg config.Config, store storage.Store, client *Client, blockLog *storage.BlockLog, checkpointMgr *checkpoint.Manager) *Service {
@@ -54,6 +65,13 @@ func NewService(cfg config.Config, store storage.Store, client *Client, blockLog
 		engine:      state.New(store, cfg.Mode),
 		checkpoints: checkpointMgr,
 		blockLog:    blockLog,
+		startedAt:   time.Now(),
+	}
+	if didCount, err := countStates(store); err == nil {
+		atomic.StoreUint64(&s.stats.DIDCount, didCount)
+	}
+	if cp, ok, err := store.GetLatestCheckpoint(); err == nil && ok {
+		atomic.StoreUint64(&s.stats.CheckpointSeq, cp.Sequence)
 	}
 	if cfg.Mode == config.ModeMirror && blockLog != nil {
 		s.appender = newBlockAppender(blockLog, 1024)
@@ -119,14 +137,22 @@ func (s *Service) RunOnce(ctx context.Context) (bool, error) {
 		return false, err
 	}
 	if len(records) == 0 {
+		atomic.StoreUint64(&s.stats.LagOps, 0)
 		return false, nil
 	}
+	latestSourceSeq := records[len(records)-1].Seq
 	committed, err := s.processRecords(ctx, records)
 	s.runOps += committed
 	if err != nil {
 		atomic.AddUint64(&s.stats.Errors, 1)
 		return committed > 0, err
 	}
+	lastCommitted := atomic.LoadUint64(&s.stats.LastSeq)
+	if latestSourceSeq > lastCommitted {
+		atomic.StoreUint64(&s.stats.LagOps, latestSourceSeq-lastCommitted)
+	} else {
+		atomic.StoreUint64(&s.stats.LagOps, 0)
+	}
 	return true, nil
 }
 
@@ -136,10 +162,11 @@ type verifyTask struct {
 }
 
 type verifyResult struct {
-	index int
-	op    types.ParsedOperation
-	state types.StateV1
-	err   error
+	index  int
+	op     types.ParsedOperation
+	state  types.StateV1
+	newDID bool
+	err    error
 }
 
 func (s *Service) processRecords(ctx context.Context, records []types.ExportRecord) (uint64, error) {
@@ -178,6 +205,7 @@ func (s *Service) verifyRecords(ctx context.Context, records []types.ExportRecor
 					continue
 				}
 				if err := s.verifier.VerifyOperation(op, existing); err != nil {
+					atomic.AddUint64(&s.stats.VerifyFailures, 1)
 					results <- verifyResult{index: task.index, err: err}
 					continue
 				}
@@ -187,7 +215,7 @@ func (s *Service) verifyRecords(ctx context.Context, records []types.ExportRecor
 					continue
 				}
 				cache[op.DID] = cloneState(&next)
-				results <- verifyResult{index: task.index, op: op, state: next}
+				results <- verifyResult{index: task.index, op: op, state: next, newDID: existing == nil}
 			}
 		}(queues[i])
 	}
@@ -287,6 +315,7 @@ func (s *Service) commitVerified(ctx context.Context, verified []verifyResult) (
 	pendingOps := make([]storage.OperationMutation, 0, batchSize)
 	pendingSeqs := make([]uint64, 0, batchSize)
 	pendingBlockHashes := map[uint64]string{}
+	pendingNewDIDs := map[string]struct{}{}
 	var committed uint64
 
 	commit := func() error {
@@ -314,6 +343,7 @@ func (s *Service) commitVerified(ctx context.Context, verified []verifyResult) (
 		lastSeq := pendingSeqs[len(pendingSeqs)-1]
 		atomic.StoreUint64(&s.stats.LastSeq, lastSeq)
 		atomic.AddUint64(&s.stats.IngestedOps, uint64(len(pendingOps)))
+		atomic.AddUint64(&s.stats.DIDCount, uint64(len(pendingNewDIDs)))
 		committed += uint64(len(pendingOps))
 
 		if s.checkpoints != nil {
@@ -329,6 +359,7 @@ func (s *Service) commitVerified(ctx context.Context, verified []verifyResult) (
 		pendingOps = pendingOps[:0]
 		pendingSeqs = pendingSeqs[:0]
 		clear(pendingBlockHashes)
+		clear(pendingNewDIDs)
 		return nil
 	}
 
@@ -350,6 +381,9 @@ func (s *Service) commitVerified(ctx context.Context, verified []verifyResult) (
 		}
 		pendingOps = append(pendingOps, storage.OperationMutation{State: v.state, Ref: ref})
 		pendingSeqs = append(pendingSeqs, v.op.Sequence)
+		if v.newDID {
+			pendingNewDIDs[v.op.DID] = struct{}{}
+		}
 		if len(pendingOps) >= batchSize {
 			if err := commit(); err != nil {
 				return committed, err
@@ -387,6 +421,10 @@ func (s *Service) createCheckpoint(sequence uint64) error {
 		float64(usage.RSSPeakKB)/1024.0,
 		time.Now().UnixMilli(),
 	)
+	atomic.StoreUint64(&s.stats.CheckpointSeq, cp.Sequence)
+	if s.metricsSink != nil {
+		s.metricsSink.ObserveCheckpoint(metrics.Total, cp.Sequence)
+	}
 	return nil
 }
 
@@ -411,10 +449,14 @@ func (s *Service) Snapshot(ctx context.Context) (types.CheckpointV1, error) {
 	if err != nil {
 		return types.CheckpointV1{}, err
 	}
-	cp, err := s.checkpoints.BuildAndStoreFromStore(seq, blocks)
+	cp, metrics, err := s.checkpoints.BuildAndStoreFromStoreWithMetrics(seq, blocks)
 	if err != nil {
 		return types.CheckpointV1{}, err
 	}
+	atomic.StoreUint64(&s.stats.CheckpointSeq, cp.Sequence)
+	if s.metricsSink != nil {
+		s.metricsSink.ObserveCheckpoint(metrics.Total, cp.Sequence)
+	}
 	return cp, nil
 }
 
@@ -473,7 +515,6 @@ func processMetrics(pid int) (cpuPct float64, rssKB int64, err error) {
 }
 
 func (s *Service) VerifyDID(ctx context.Context, did string) error {
-	_ = ctx
 	if err := s.CorruptionError(); err != nil {
 		return err
 	}
@@ -497,6 +538,9 @@ func (s *Service) VerifyDID(ctx context.Context, did string) error {
 
 	var previous *types.StateV1
 	for _, seq := range seqs {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
 		ref, ok, err := s.store.GetOpSeqRef(seq)
 		if err != nil {
 			return err
@@ -523,7 +567,6 @@ func (s *Service) VerifyDID(ctx context.Context, did string) error {
 }
 
 func (s *Service) RecomputeTipAtOrBefore(ctx context.Context, did string, sequence uint64) (string, []uint64, error) {
-	_ = ctx
 	if err := s.CorruptionError(); err != nil {
 		return "", nil, err
 	}
@@ -551,6 +594,9 @@ func (s *Service) RecomputeTipAtOrBefore(ctx context.Context, did string, sequen
 	var previous *types.StateV1
 	var tip string
 	for _, seq := range filtered {
+		if err := ctx.Err(); err != nil {
+			return "", nil, err
+		}
 		ref, ok, err := s.store.GetOpSeqRef(seq)
 		if err != nil {
 			return "", nil, err
@@ -669,12 +715,37 @@ func (s *Service) Close() {
 	}
 }
 
+func (s *Service) SetMetricsSink(sink MetricsSink) {
+	s.metricsSink = sink
+}
+
 func (s *Service) Stats() Stats {
+	ingested := atomic.LoadUint64(&s.stats.IngestedOps)
+	opsPerSec := 0.0
+	if elapsed := time.Since(s.startedAt).Seconds(); elapsed > 0 {
+		opsPerSec = float64(ingested) / elapsed
+	}
 	return Stats{
-		IngestedOps: atomic.LoadUint64(&s.stats.IngestedOps),
-		Errors:      atomic.LoadUint64(&s.stats.Errors),
-		LastSeq:     atomic.LoadUint64(&s.stats.LastSeq),
+		IngestedOps:     ingested,
+		Errors:          atomic.LoadUint64(&s.stats.Errors),
+		LastSeq:         atomic.LoadUint64(&s.stats.LastSeq),
+		VerifyFailures:  atomic.LoadUint64(&s.stats.VerifyFailures),
+		LagOps:          atomic.LoadUint64(&s.stats.LagOps),
+		DIDCount:        atomic.LoadUint64(&s.stats.DIDCount),
+		CheckpointSeq:   atomic.LoadUint64(&s.stats.CheckpointSeq),
+		IngestOpsPerSec: opsPerSec,
+	}
+}
+
+func countStates(store storage.Store) (uint64, error) {
+	var count uint64
+	if err := store.ForEachState(func(types.StateV1) error {
+		count++
+		return nil
+	}); err != nil {
+		return 0, err
 	}
+	return count, nil
 }
 
 type appendResult struct {
diff --git a/internal/ingest/service_integration_test.go b/internal/ingest/service_integration_test.go
index d7bd1b6..e01dd4e 100644
--- a/internal/ingest/service_integration_test.go
+++ b/internal/ingest/service_integration_test.go
@@ -155,3 +155,72 @@ func buildSignedRecords(t *testing.T) []types.ExportRecord {
 	out = append(out, rec1, rec2, rec3)
 	return out
 }
+
+func TestRecomputeTipAtOrBeforeHonorsContextCancellation(t *testing.T) {
+	tmp := t.TempDir()
+	dataDir := filepath.Join(tmp, "data")
+	if err := os.MkdirAll(dataDir, 0o755); err != nil {
+		t.Fatalf("mkdir data dir: %v", err)
+	}
+
+	keySeed := make([]byte, ed25519.SeedSize)
+	if _, err := rand.Read(keySeed); err != nil {
+		t.Fatalf("rand seed: %v", err)
+	}
+	keyPath := filepath.Join(tmp, "mirror.key")
+	if err := os.WriteFile(keyPath, []byte(base64.RawURLEncoding.EncodeToString(keySeed)), 0o600); err != nil {
+		t.Fatalf("write mirror key: %v", err)
+	}
+
+	records := buildSignedRecords(t)
+	sourcePath := filepath.Join(tmp, "sample.ndjson")
+	file, err := os.Create(sourcePath)
+	if err != nil {
+		t.Fatalf("create source: %v", err)
+	}
+	for _, rec := range records {
+		b, _ := json.Marshal(rec)
+		if _, err := fmt.Fprintln(file, string(b)); err != nil {
+			t.Fatalf("write source: %v", err)
+		}
+	}
+	file.Close()
+
+	store, err := storage.OpenPebble(dataDir)
+	if err != nil {
+		t.Fatalf("open pebble: %v", err)
+	}
+	defer store.Close()
+	if err := store.SetMode(config.ModeMirror); err != nil {
+		t.Fatalf("set mode: %v", err)
+	}
+	bl, err := storage.OpenBlockLog(dataDir, 3, 4)
+	if err != nil {
+		t.Fatalf("open block log: %v", err)
+	}
+	cfg := config.Config{
+		Mode:                 config.ModeMirror,
+		DataDir:              dataDir,
+		PLCSource:            sourcePath,
+		VerifyPolicy:         config.VerifyFull,
+		ZstdLevel:            3,
+		BlockSizeMB:          4,
+		CheckpointInterval:   2,
+		ListenAddr:           ":0",
+		MirrorPrivateKeyPath: keyPath,
+		PollInterval:         5,
+	}
+	service := NewService(cfg, store, NewClient(sourcePath), bl, checkpoint.NewManager(store, dataDir, keyPath))
+	if err := service.Replay(context.Background()); err != nil {
+		t.Fatalf("replay: %v", err)
+	}
+	if err := service.Flush(context.Background()); err != nil {
+		t.Fatalf("flush: %v", err)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, _, err := service.RecomputeTipAtOrBefore(ctx, "did:plc:alice", 2); err == nil {
+		t.Fatalf("expected cancellation error")
+	}
+}