tap: add blocked handler diagnostics

.agents/skills/hyperindex/SKILL.md

@@ -42,6 +42,17 @@ The current production indexer is ATProto-first. Do **not** explain it using old
 
 Use production by default for consumer examples. `api.indexer.hypercerts.dev` is the production endpoint and currently exposes presence filters and typed queries for the collections above. Staging and production do not always run the same schema, and this skill usually reflects the default branch before every endpoint has caught up. Because ATProto data is network-wide, do not describe staging as a separate dataset unless you have verified an environment-specific indexing difference.
 
+## Operational diagnostics
+
+Use the HTTP `/stats` endpoint on the target host, without `/graphql`, when an operator needs to diagnose Tap ingestion. Deployments with phase-aware Tap diagnostics expose these non-sensitive fields under `tap`:
+
+- `last_event_received_at` — timestamp of the most recently dispatched Tap event.
+- `last_ack_at` — timestamp of the most recent successful Tap acknowledgement.
+- `in_flight` — optional current event metadata, processing phase, start time, and duration. It is absent when no event is being processed.
+- `database_pool` — connection counts plus cumulative wait count and wait duration.
+
+For example, use `https://api.indexer.hypercerts.dev/stats` for production or `https://dev.api.indexer.hypercerts.dev/stats` for staging. Confirm the target deployment includes these fields before relying on them; hosted environments may run different Hyperindex revisions. The endpoint does not expose record bodies or credentials.
+
 ## Before answering
 
 1. If the user asks for an exact field, filter, enum, or union and you are not sure, introspect the endpoint first.

.agents/skills/hyperindex/references/schema-reference.md

@@ -1,12 +1,23 @@
 # Hyperindex GraphQL Schema Reference
 
-Last updated on 2026-07-28 for pending record author identity changes. Baseline generated from live introspection of `https://api.indexer.hypercerts.dev/graphql` on 2026-06-10; `author` metadata described below reflects the current branch and may not yet be deployed.
+Last updated on 2026-08-24 for pending Tap operational diagnostics and record author identity changes. Baseline generated from live introspection of `https://api.indexer.hypercerts.dev/graphql` on 2026-06-10; current-branch behavior described below may not yet be deployed.
 
 ## Endpoints
 
 - Production GraphQL: `https://api.indexer.hypercerts.dev/graphql`
 - Staging GraphQL: `https://dev.api.indexer.hypercerts.dev/graphql`
 
+## Operational HTTP diagnostics
+
+Deployments with phase-aware Tap diagnostics expose non-sensitive ingestion state at `/stats` on the same host:
+
+- `tap.last_event_received_at` — most recently dispatched Tap event timestamp.
+- `tap.last_ack_at` — most recent successful Tap acknowledgement timestamp.
+- `tap.in_flight` — optional current event object containing `event_id`, `type`, `did`, `collection`, `rkey`, `action`, `phase`, `started_at`, and `duration_ms`; absent when no event is being processed.
+- `tap.database_pool` — `max_open_connections`, `open_connections`, `in_use`, `idle`, `wait_count`, and `wait_duration_ms`.
+
+Use `https://api.indexer.hypercerts.dev/stats` for production and `https://dev.api.indexer.hypercerts.dev/stats` for staging. Confirm field availability on the target deployment because hosted environments may run different revisions. These diagnostics do not expose record bodies or credentials.
+
 ## Query fields
 
 | Query | Arguments | Purpose |

.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
 }