tap: add blocked handler diagnostics

.changes/unreleased/add-tap-handler-diagnostics.yaml

@@ -0,0 +1,4 @@
+kind: added
+body: Add phase-aware Tap ingestion diagnostics for operators. Events that remain in a handler for five seconds now produce recurring warnings with their current operation and database pool state, while `/stats` exposes last-received and last-acknowledged timestamps plus any in-flight event so ingestion stalls can be diagnosed without changing delivery behavior.
+custom:
+  Affects: operator

README.md

@@ -121,6 +121,10 @@ Tap Docker deployments also require `ADMIN_API_KEY` in `.env` because Hyperindex
 
 When Tap mode starts, Hyperindex performs a bounded background reconciliation for actor rows whose handle is missing. It reads current identity metadata from Tap's local `/info/:did` API, never delays API startup, and never replaces a handle populated concurrently by a newer identity event.
 
+Tap event processing has a phase-aware watchdog. If one event remains in its handler for five seconds, Hyperindex logs `Tap event processing is blocked` with the event ID, record identity, current operation phase, duration, and database pool state. It repeats the warning every 30 seconds until processing finishes, then logs `Slow Tap event processing completed`. Record bodies and credentials are never included.
+
+`GET /stats` exposes the same non-sensitive state under `tap`: `last_event_received_at`, `last_ack_at`, an optional `in_flight` event with its current phase and duration, and `database_pool` connection and wait counters. Use these fields to distinguish a blocked database operation from a disconnected or idle consumer without changing acknowledgement or retry behavior.
+
 **Local isolated Tap smoke stack (requires Docker):**
 
 Use this when you want to test current local changes against Tap with the Hypercerts/Certified lexicon set mounted into the backend container:

cmd/hyperindex/main.go

@@ -348,15 +348,7 @@ func setupRouter(cfg *config.Config, svc *services, bg *backgroundServices) *chi
 		}
 
 		if cfg.TapEnabled && bg.tapConsumer != nil {
-			tapStats := bg.tapConsumer.Stats()
-			tapInfo := map[string]any{
-				"events_received": tapStats.EventsReceived,
-				"records_created": tapStats.RecordsCreated,
-				"records_updated": tapStats.RecordsUpdated,
-				"records_deleted": tapStats.RecordsDeleted,
-				"identity_events": tapStats.IdentityEvents,
-				"errors":          tapStats.Errors,
-			}
+			tapInfo := tapStatsInfo(bg.tapConsumer.Stats())
 			// Include Tap sidecar health in stats for observability (non-blocking).
 			if bg.tapAdminClient != nil {
 				applyTapSidecarHealth(reqCtx, tapInfo, 1500*time.Millisecond, bg.tapAdminClient.Health)
@@ -404,6 +396,47 @@ func setupRouter(cfg *config.Config, svc *services, bg *backgroundServices) *chi
 	return r
 }
 
+func tapStatsInfo(stats tap.Stats) map[string]any {
+	info := map[string]any{
+		"events_received": stats.EventsReceived,
+		"records_created": stats.RecordsCreated,
+		"records_updated": stats.RecordsUpdated,
+		"records_deleted": stats.RecordsDeleted,
+		"identity_events": stats.IdentityEvents,
+		"errors":          stats.Errors,
+	}
+	if stats.LastEventReceivedAt != nil {
+		info["last_event_received_at"] = stats.LastEventReceivedAt.UTC().Format(time.RFC3339Nano)
+	}
+	if stats.LastAckAt != nil {
+		info["last_ack_at"] = stats.LastAckAt.UTC().Format(time.RFC3339Nano)
+	}
+	if event := stats.InFlight; event != nil {
+		info["in_flight"] = map[string]any{
+			"event_id":    event.EventID,
+			"type":        event.Type,
+			"did":         event.DID,
+			"collection":  event.Collection,
+			"rkey":        event.RKey,
+			"action":      event.Action,
+			"phase":       event.Phase,
+			"started_at":  event.StartedAt.UTC().Format(time.RFC3339Nano),
+			"duration_ms": event.Duration.Milliseconds(),
+		}
+	}
+	if pool := stats.DatabasePool; pool != nil {
+		info["database_pool"] = map[string]any{
+			"max_open_connections": pool.MaxOpenConnections,
+			"open_connections":     pool.OpenConnections,
+			"in_use":               pool.InUse,
+			"idle":                 pool.Idle,
+			"wait_count":           pool.WaitCount,
+			"wait_duration_ms":     pool.WaitDuration.Milliseconds(),
+		}
+	}
+	return info
+}
+
 func databaseReady(ctx context.Context, svc *services) error {
 	if svc == nil || svc.db == nil {
 		return errors.New("database executor is unavailable")
@@ -971,6 +1004,17 @@ func startTap(
 		TapURL:      tapURL,
 		Password:    cfg.TapAdminPassword,
 		DisableAcks: cfg.TapDisableAcks,
+		DatabaseStats: func() tap.DatabasePoolStats {
+			stats := svc.db.DB().Stats()
+			return tap.DatabasePoolStats{
+				MaxOpenConnections: stats.MaxOpenConnections,
+				OpenConnections:    stats.OpenConnections,
+				InUse:              stats.InUse,
+				Idle:               stats.Idle,
+				WaitCount:          stats.WaitCount,
+				WaitDuration:       stats.WaitDuration,
+			}
+		},
 	}, handler)
 
 	// Store consumer reference for clean shutdown.

cmd/hyperindex/main_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/GainForest/hyperindex/internal/buildinfo"
 	"github.com/GainForest/hyperindex/internal/config"
 	"github.com/GainForest/hyperindex/internal/database/repositories"
+	"github.com/GainForest/hyperindex/internal/tap"
 	"github.com/GainForest/hyperindex/internal/testutil"
 )
 
@@ -318,6 +319,63 @@ func labelerTestServices(db *testutil.TestDB) *services {
 	}
 }
 
+func TestTapStatsInfoIncludesProcessingAndDatabaseDiagnostics(t *testing.T) {
+	startedAt := time.Date(2026, 8, 22, 1, 2, 3, 4, time.UTC)
+	lastReceived := startedAt.Add(-time.Second)
+	lastAck := startedAt.Add(-2 * time.Second)
+
+	got := tapStatsInfo(tap.Stats{
+		EventsReceived:      12,
+		RecordsCreated:      3,
+		RecordsUpdated:      2,
+		RecordsDeleted:      1,
+		IdentityEvents:      4,
+		Errors:              5,
+		LastEventReceivedAt: &lastReceived,
+		LastAckAt:           &lastAck,
+		InFlight: &tap.InFlightEventStats{
+			EventID:    99,
+			Type:       tap.EventTypeRecord,
+			DID:        "did:plc:blocked",
+			Collection: "org.hypercerts.claim.activity",
+			RKey:       "slow-record",
+			Action:     tap.ActionCreate,
+			Phase:      "records.insert",
+			StartedAt:  startedAt,
+			Duration:   1500 * time.Millisecond,
+		},
+		DatabasePool: &tap.DatabasePoolStats{
+			MaxOpenConnections: 25,
+			OpenConnections:    25,
+			InUse:              25,
+			Idle:               0,
+			WaitCount:          7,
+			WaitDuration:       3 * time.Second,
+		},
+	})
+
+	if got["last_event_received_at"] != lastReceived.Format(time.RFC3339Nano) {
+		t.Errorf("last_event_received_at = %#v", got["last_event_received_at"])
+	}
+	if got["last_ack_at"] != lastAck.Format(time.RFC3339Nano) {
+		t.Errorf("last_ack_at = %#v", got["last_ack_at"])
+	}
+	inFlight, ok := got["in_flight"].(map[string]any)
+	if !ok {
+		t.Fatalf("in_flight = %#v, want object", got["in_flight"])
+	}
+	if inFlight["event_id"] != int64(99) || inFlight["phase"] != "records.insert" || inFlight["duration_ms"] != int64(1500) {
+		t.Errorf("in_flight = %#v", inFlight)
+	}
+	pool, ok := got["database_pool"].(map[string]any)
+	if !ok {
+		t.Fatalf("database_pool = %#v, want object", got["database_pool"])
+	}
+	if pool["in_use"] != 25 || pool["wait_count"] != int64(7) || pool["wait_duration_ms"] != int64(3000) {
+		t.Errorf("database_pool = %#v", pool)
+	}
+}
+
 func TestApplyTapSidecarHealth(t *testing.T) {
 	tests := []struct {
 		name      string

internal/tap/consumer.go

@@ -24,6 +24,14 @@ const (
 	// defaultReadTimeout is the timeout for WebSocket read operations.
 	defaultReadTimeout = 60 * time.Second
 
+	// defaultSlowEventThreshold is how long an event may remain in one handler
+	// before the consumer starts emitting blocked-processing diagnostics.
+	defaultSlowEventThreshold = 5 * time.Second
+
+	// defaultBlockedEventLogInterval controls follow-up diagnostics while the
+	// same event remains in flight.
+	defaultBlockedEventLogInterval = 30 * time.Second
+
 	// minBackoff is the initial reconnection backoff duration.
 	minBackoff = time.Second
 
@@ -47,6 +55,14 @@ type ConsumerConfig struct {
 
 	// DisableAcks puts the consumer in fire-and-forget mode (no acks sent).
 	DisableAcks bool
+
+	// SlowEventThreshold and BlockedEventLogInterval override the production
+	// watchdog timings. Zero values use the defaults.
+	SlowEventThreshold      time.Duration
+	BlockedEventLogInterval time.Duration
+
+	// DatabaseStats supplies a non-sensitive pool snapshot for diagnostics.
+	DatabaseStats func() DatabasePoolStats
 }
 
 // EventHandler processes Tap events. Return nil to ack, error to nack.
@@ -55,14 +71,42 @@ type EventHandler interface {
 	HandleIdentity(ctx context.Context, event *IdentityEvent) error
 }
 
-// Stats tracks consumer statistics.
+// DatabasePoolStats contains the database/sql pool state useful when an event
+// is waiting on a connection or database operation.
+type DatabasePoolStats struct {
+	MaxOpenConnections int
+	OpenConnections    int
+	InUse              int
+	Idle               int
+	WaitCount          int64
+	WaitDuration       time.Duration
+}
+
+// InFlightEventStats describes the single Tap event currently being processed.
+type InFlightEventStats struct {
+	EventID    int64
+	Type       EventType
+	DID        string
+	Collection string
+	RKey       string
+	Action     ActionType
+	Phase      string
+	StartedAt  time.Time
+	Duration   time.Duration
+}
+
+// Stats tracks consumer statistics and processing diagnostics.
 type Stats struct {
-	EventsReceived int64
-	RecordsCreated int64
-	RecordsUpdated int64
-	RecordsDeleted int64
-	IdentityEvents int64
-	Errors         int64
+	EventsReceived      int64
+	RecordsCreated      int64
+	RecordsUpdated      int64
+	RecordsDeleted      int64
+	IdentityEvents      int64
+	Errors              int64
+	LastEventReceivedAt *time.Time
+	LastAckAt           *time.Time
+	InFlight            *InFlightEventStats
+	DatabasePool        *DatabasePoolStats
 }
 
 // Consumer connects to Tap's WebSocket and dispatches events.
@@ -91,6 +135,11 @@ type Consumer struct {
 	recordsDeleted int64
 	identityEvents int64
 	errors         int64
+
+	diagnosticsMu       sync.RWMutex
+	lastEventReceivedAt time.Time
+	lastAckAt           time.Time
+	inFlight            *eventTrace
 }
 
 // NewConsumer creates a new Tap consumer.
@@ -267,17 +316,22 @@ func (c *Consumer) runOnce(ctx context.Context) (bool, bool, error) {
 }
 
 // dispatch routes an event to the appropriate handler and sends an ack on success.
-func (c *Consumer) dispatch(ctx context.Context, conn *websocket.Conn, event *Event) error {
+func (c *Consumer) dispatch(ctx context.Context, conn *websocket.Conn, event *Event) (err error) {
+	ctx, trace := c.beginEvent(ctx, event)
+	defer func() { c.finishEvent(trace, err) }()
+
 	var handlerErr error
 
 	switch {
 	case event.IsRecord():
+		setEventPhase(ctx, "handler.record")
 		handlerErr = c.handler.HandleRecord(ctx, event.Record)
 		if handlerErr == nil {
 			c.incrementRecordStat(event.Record.Action)
 		}
 
 	case event.IsIdentity():
+		setEventPhase(ctx, "handler.identity")
 		handlerErr = c.handler.HandleIdentity(ctx, event.Identity)
 		if handlerErr == nil {
 			atomic.AddInt64(&c.identityEvents, 1)
@@ -297,6 +351,7 @@ func (c *Consumer) dispatch(ctx context.Context, conn *websocket.Conn, event *Ev
 	// The Tap server expects JSON: {"type":"ack","id":<id>}
 	// See: https://github.com/bluesky-social/indigo/blob/main/cmd/tap/types.go
 	if !c.config.DisableAcks {
+		setEventPhase(ctx, "ack.write")
 		ackMsg := fmt.Sprintf(`{"type":"ack","id":%d}`, event.ID)
 		writeFn := c.writeText
 		if c.writeTextFn != nil {
@@ -305,8 +360,10 @@ func (c *Consumer) dispatch(ctx context.Context, conn *websocket.Conn, event *Ev
 		if err := writeFn(conn, ackMsg); err != nil {
 			return fmt.Errorf("failed to send ack for event %d: %w", event.ID, err)
 		}
+		c.markAcked(time.Now())
 	}
 
+	setEventPhase(ctx, "completed")
 	return nil
 }
 
@@ -358,14 +415,33 @@ func (c *Consumer) Stop() {
 	})
 }
 
-// Stats returns the current event counts.
+// Stats returns the current event counts and processing diagnostics.
 func (c *Consumer) Stats() Stats {
-	return Stats{
+	stats := Stats{
 		EventsReceived: atomic.LoadInt64(&c.eventsReceived),
 		RecordsCreated: atomic.LoadInt64(&c.recordsCreated),
 		RecordsUpdated: atomic.LoadInt64(&c.recordsUpdated),
 		RecordsDeleted: atomic.LoadInt64(&c.recordsDeleted),
 		IdentityEvents: atomic.LoadInt64(&c.identityEvents),
 		Errors:         atomic.LoadInt64(&c.errors),
 	}
+
+	c.diagnosticsMu.RLock()
+	if !c.lastEventReceivedAt.IsZero() {
+		lastReceived := c.lastEventReceivedAt
+		stats.LastEventReceivedAt = &lastReceived
+	}
+	if !c.lastAckAt.IsZero() {
+		lastAck := c.lastAckAt
+		stats.LastAckAt = &lastAck
+	}
+	if c.inFlight != nil {
+		stats.InFlight = c.inFlight.snapshot(time.Now())
+	}
+	c.diagnosticsMu.RUnlock()
+	if c.config.DatabaseStats != nil {
+		pool := c.config.DatabaseStats()
+		stats.DatabasePool = &pool
+	}
+	return stats
 }