[PROPOSAL] Resilient PIT queries

[sean-]

Problem

OpenSearch Point-In-Time (PIT) search contexts are pinned to specific (node, shard, contextId) tuples encoded in the PIT id. When a node restarts (e.g., during an automated rolling kernel upgrade), every search context held on that node is gone. There is no server-side replica fallback for PIT searches: once a PIT id is established, queries execute against the exact contexts encoded in the id, and a missing context surfaces as a shard failure.

For deployments with mandatory rolling restart schedules, any long-running PIT-paginated workload (reindex, export, sync) that crosses a restart window fails or returns silently incomplete results.

Why this has to live in the client

Server-side fixes (replica fallback, context migration) are not on the OpenSearch roadmap and would change PIT's snapshot semantics. The fix has to live in the client.

The fix is conceptually simple: when a PIT is invalidated, recreate it and resume from the last search_after cursor. The cost is a narrow window of weakened snapshot semantics around the restart boundary. For pagination, ETL, and reindex-style workloads this is acceptable; for strict snapshot reads it is not, so the resilient path must be explicit and observable.

Failure modes (server response shapes)

Three shapes matter for client-side classification:

1. Partial shard failure (most common during rolling restart). HTTP 200, _shards.failed > 0, _shards.failures[].reason.type == "search_context_missing_exception". Other shards' hits are still in hits.hits. This is what you see when one node out of N is mid-restart.
2. Total context loss. HTTP 404, top-level error.type == "search_context_missing_exception". Every referenced node has lost its contexts.
3. Transport / 5xx. Connection reset, node-gone, etc. PIT state is unknown until probed (size=0 against the PIT id).

Server source references:

• server/src/main/java/org/opensearch/search/SearchContextMissingException.java
• server/src/main/java/org/opensearch/action/search/SearchContextId.java
• libs/core/src/main/java/org/opensearch/core/rest/RestStatus.java (HTTP status aggregation across shards)

Proposed design

A Paginator on the PIT subclient that owns PIT lifecycle and transparently recreates the PIT on classified failures, resuming via search_after.

Versioning

• v4 (opensearchapi): API surface added; resilience knobs ship zero-valued. No automatic recreates unless the caller opts in. Forward-compatible experimentation path; today's behavior unchanged.
• v5preview (v5preview/opensearchapi): PIT methods relocated from top-level Client onto a client.PIT subclient (matching v4's shape). Paginator ships with sane defaults: keep-alive 5m, max recreate attempts 3, capped exponential backoff, hit-id dedup window 32768.

Iterator surface (Go 1.23+ iter.Seq2)

The paginator iterates over hits or pages depending on configuration. Hit-level iteration is the default because it enables _id-based deduplication across snapshot breaks; page-level iteration is offered for callers that batch per-page.

p, err := client.PIT.Paginator(ctx, PaginateReq{
    Indices: []string{"my-index"},
    Body:    body, // must include a deterministic Sort
})
if err != nil { /* ... */ }
defer p.Close(ctx)

// hit-level (recommended)
for hit, err := range p.Hits(ctx) {
    switch {
    case errors.Is(err, pit.ErrHitOutOfOrder):
        // hit is still valid; surface to ops, then process or skip
        metrics.OutOfOrder.Inc()
        process(hit)
    case err != nil:
        return err // fatal
    default:
        process(hit)
    }
}

// or page-level
for page, err := range p.Pages(ctx) {
    switch {
    case errors.Is(err, pit.ErrSnapshotBreak):
        // page is valid; previously-paginated pages may be stale relative
        // to the new snapshot. Caller decides: log, restart, or mark partial.
        log.Warn("snapshot break", "old_pit", page.PriorPITID, "new_pit", p.PITID())
        batchWrite(page.Hits.Hits)
    case err != nil:
        return err
    default:
        batchWrite(page.Hits.Hits)
    }
}

Snapshot-break semantics

Across a recreate boundary the paginator cannot guarantee strict ordering or exactly-once delivery. Two sentinel errors surface the anomaly without forcing the caller to abort:

• ErrSnapshotBreak (Pages). Returned alongside the first page emitted from a recreated PIT. Signals: "previously-paginated pages may be stale; the new snapshot includes writes since the old PIT was created, and any of those writes that would belong on a page already past the cursor will not be visible to this paginator." Caller decides: log, restart from scratch, or mark the workload partial downstream.
• ErrHitOutOfOrder (Hits). Returned alongside a hit that was already seen by _id in a prior PIT but reappears with a different version after a recreate. The hit value is the newer version. Caller decides: process the new version, skip, or surface to ops.
• ErrShardFailure (Hits and Pages). Returned alongside a hit/page that came from a search response with _shards.failed > 0 for a non-PIT-context reason (circuit breaker, task cancelled, etc.). The cursor has advanced past the failed shard's slice, so docs from that shard at this cursor range are lost. Caller decides: tolerate (mark partial) or abort. The paginator does not retry or recreate, because PIT pins to specific contexts and a non-context shard failure is not a snapshot problem.

Note: both errors are returned with a valid V value. This is a deliberate departure from "err != nil means V is zero" and is documented in the package doc. Fatal errors (context cancellation, exhausted recreate budget, transport failure that does not classify) still return a zero V.

The caller picks one of two delivery modes:

• Skip-on-dedup (default for Hits). Maintain a bounded LRU set of recently-emitted _ids and skip duplicates after a recreate, surfacing ErrHitOutOfOrder for the version-changed case. Trade: may omit documents whose sort values shifted across the boundary and now fall behind the cursor.
• Emit-all. Disable dedup. Trade: may emit duplicates and out-of-order documents around the boundary.

Pages cannot dedup individual hits and inherits emit-all by default; only ErrSnapshotBreak applies.

Observability

Two layers, matching the rest of the client:

Counters and gauges via the existing client.Metrics() snapshot. The PIT paginator increments fields on the internal opensearchtransport.metrics struct; new fields on the public Metrics snapshot:

// added to opensearchtransport.Metrics
PITsOpened       int            `json:"pits_opened"`        // counter
PITsClosed       int            `json:"pits_closed"`        // counter
PITsRecreated    int            `json:"pits_recreated"`     // counter
PITHitsEmitted   int64          `json:"pit_hits_emitted"`   // counter
PITPagesEmitted  int64          `json:"pit_pages_emitted"`  // counter
PITShardFailures map[string]int `json:"pit_shard_failures"` // counter, keyed by reason.type

The "currently open PITs" gauge is PITsOpened - PITsClosed, computed at scrape time. PITShardFailures is keyed by reason.type (e.g., search_context_missing_exception, circuit_breaking_exception) so callers can split PIT-related from transient. Callers wire client.Metrics() into Prom/OTel via their own existing scraper; we do not take a dependency on either library.

Per-event callback for rich context. OnSnapshotBreak (already on PITPaginatorConfig) fires once per recreate with old/new PIT id, cursor, failed-shards detail, cause, and attempt number. This is the hook ops uses when a counter spike needs to be cross-referenced with a specific incident. Counters alone do not carry that context.

Configuration

type PITPaginatorConfig struct {
    KeepAlive           string                            // v5 default "5m"; v4 required
    MaxRecreateAttempts int                               // v5 default 3; v4 default 0
    Backoff             func(attempt int) time.Duration  // v5 default capped exponential; v4 default nil
    SeenIDsWindow       int                               // v5 default 32768; v4 default 0
    OnSnapshotBreak     func(SnapshotBreakEvent)          // nil-safe
}

type SnapshotBreakEvent struct {
    OldPITID     string
    NewPITID     string
    Cursor       []any
    FailedShards []ShardSearchFailure
    Cause        error
    Attempt      int
}

Memory footprint for the dedup window. A bounded LRU implemented as map[string]struct{} plus a circular buffer of _id strings (cheaper than container/list, same API surface). Per-entry cost on 64-bit Go: ~16 B string header + ~20 B string data (default OpenSearch auto-generated _id is 20 chars; see TimeBasedUUIDGenerator.java, 15-byte UUID base64-encoded without padding) + ~32 B amortized map slot + ~16 B ring slot ≈ ~75-80 B per entry. At the v5 default of 32768 entries, steady-state is roughly 2.4 MiB per fully-loaded paginator.

Custom _id schemes shift this:

pattern	length	est. per entry	32K total
OpenSearch auto-gen	20	~75 B	~2.4 MiB
ULID	26	~82 B	~2.6 MiB
RFC 4122 UUID string	36	~92 B	~2.9 MiB
Composite keys (timestamp + source + key)	60-100	~110-150 B	~3.5-5 MiB

Adjust SeenIDsWindow for high-cardinality jobs, long custom _ids, or memory-constrained processes. A bloom filter would lower the per-entry cost (and admit false positives, which here means rare missed dedups, i.e., emitting a duplicate the caller could have skipped); deferred as an optimization until profiling justifies it.

Lifecycle

The paginator creates and deletes the PIT by default. WithExistingPIT(id) is an escape hatch for callers sharing a PIT across runs; in that mode the paginator does not delete on Close.

Out of scope

• Request hedging / latency-driven retries. Caller owns ctx; timeout-driven concurrency is the caller's responsibility.
• Server-side changes (replica fallback, context migration).
• Caller-supplied error classifier. Classification is internal until a concrete use case justifies exposing it.

Open questions

• Default backoff. Proposing capped exponential with full jitter: delay = jitter(min(250ms * 2^attempt, 10s)). Rationale: PIT recreate is not retrying a hot query against an overloaded cluster, it's recreating a fresh id that picks up surviving replicas, so the backoff exists to absorb classifier flakiness (e.g., a probe race between the old PIT id partially expiring) rather than to wait out a node restart. Three attempts with this curve totals at most ~14s of wall time before giving up. Tunable via Backoff func(attempt int) time.Duration for callers with stricter SLOs.
• WithExistingPIT + MaxRecreateAttempts > 0 interaction. If the caller hands us a PIT id and we recreate it on failure, the new id is not the one they manage. Options: (a) refuse the combination at construction time, (b) recreate freely and surface the new id via OnSnapshotBreak so the caller can update their bookkeeping, (c) recreate but mark the paginator detached from the original id. Leaning toward (b); flagging for review.