fix: Keep search results at a resource's current state

cmd/search/indexer/command.go

@@ -29,6 +29,12 @@ import (
 // process, so it is duplicated here to bound the indexing timeouts.
 const consumerAckWait = 300 * time.Second
 
+// docStateTTLDefault mirrors the audit stream's maxAge in
+// config/components/nats-streams/nats-streams.yaml. A record must outlive
+// every event that could still be redelivered for its document, or a late
+// event could be applied with nothing left to order it against.
+const docStateTTLDefault = 7 * 24 * time.Hour
+
 const (
 	// heartbeatMargin is how many heartbeat intervals an ackWait must fit. The
 	// spare intervals absorb a lost or delayed tick without JetStream giving up on
@@ -67,6 +73,12 @@ type ResourceIndexerOptions struct {
 	NatsReindexStream       string
 	NatsReindexConsumerName string
 
+	// Document state bucket, shared by every replica so that writes to one
+	// document stay in resourceVersion order whichever replica flushes last.
+	NatsDocStateBucket   string
+	NatsDocStateReplicas int
+	DocStateTTL          time.Duration
+
 	// Meilisearch connection and timeout settings
 	MeilisearchTaskWaitTimeout  time.Duration
 	MeilisearchTaskPollInterval time.Duration
@@ -98,6 +110,9 @@ func NewResourceIndexerOptions() *ResourceIndexerOptions {
 		NatsStreamName:              "AUDIT_EVENTS",
 		NatsReindexStream:           "REINDEX_EVENTS",
 		NatsReindexConsumerName:     "search-reindexer",
+		NatsDocStateBucket:          "SEARCH_INDEXER_DOC_STATE",
+		NatsDocStateReplicas:        1,
+		DocStateTTL:                 docStateTTLDefault,
 		MeilisearchTaskWaitTimeout:  10 * time.Minute,
 		MeilisearchTaskPollInterval: 500 * time.Millisecond,
 		MeilisearchHTTPTimeout:      60 * time.Second,
@@ -123,6 +138,9 @@ func (o *ResourceIndexerOptions) AddFlags(fs *pflag.FlagSet) {
 
 	fs.StringVar(&o.NatsReindexStream, "nats-reindex-stream", o.NatsReindexStream, "The JetStream stream name for re-index messages.")
 	fs.StringVar(&o.NatsReindexConsumerName, "nats-reindex-consumer-name", o.NatsReindexConsumerName, "The name of the re-index JetStream consumer (must match the manifest).")
+	fs.StringVar(&o.NatsDocStateBucket, "nats-doc-state-bucket", o.NatsDocStateBucket, "The JetStream KV bucket that records the newest indexed state of every document. It is created if missing and must be shared by all replicas.")
+	fs.IntVar(&o.NatsDocStateReplicas, "nats-doc-state-replicas", o.NatsDocStateReplicas, "The number of JetStream replicas for the document state bucket.")
+	fs.DurationVar(&o.DocStateTTL, "doc-state-ttl", o.DocStateTTL, "How long a document's recorded state is kept after its last write. Must cover the audit stream's retention so no redelivered event can outlive the record that orders it.")
 	fs.StringVar(&o.NatsTLSCA, "nats-tls-ca", o.NatsTLSCA, "The path to the NATS TLS CA file.")
 	fs.StringVar(&o.NatsTLSCert, "nats-tls-cert", o.NatsTLSCert, "The path to the NATS TLS certificate file.")
 	fs.StringVar(&o.NatsTLSKey, "nats-tls-key", o.NatsTLSKey, "The path to the NATS TLS key file.")
@@ -168,6 +186,15 @@ func (o *ResourceIndexerOptions) Validate() error {
 	if o.NatsReindexConsumerName == "" {
 		return fmt.Errorf("nats-reindex-consumer-name must be set")
 	}
+	if o.NatsDocStateBucket == "" {
+		return fmt.Errorf("nats-doc-state-bucket must be set")
+	}
+	if o.NatsDocStateReplicas < 1 {
+		return fmt.Errorf("nats-doc-state-replicas must be at least 1")
+	}
+	if o.DocStateTTL < time.Hour {
+		return fmt.Errorf("doc-state-ttl must be at least 1h")
+	}
 	if o.MeilisearchDomain == "" {
 		return fmt.Errorf("meilisearch-domain must be set")
 	}
@@ -400,12 +427,28 @@ func Run(o *ResourceIndexerOptions, ctx context.Context) error {
 		return fmt.Errorf("failed to create search client: %w", err)
 	}
 
+	// The bucket is owned by the indexer rather than declared alongside the
+	// consumers because every replica, and the re-index path, must agree on
+	// the one arbiter for a document's newest state.
+	docStateKV, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
+		Bucket:      o.NatsDocStateBucket,
+		Description: "Newest indexed state per document, shared by search indexer replicas to keep writes in resourceVersion order",
+		History:     1,
+		TTL:         o.DocStateTTL,
+		Storage:     jetstream.FileStorage,
+		Replicas:    o.NatsDocStateReplicas,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to open document state bucket %s: %w", o.NatsDocStateBucket, err)
+	}
+
 	batchConfig := indexer.BatchConfig{
 		BatchSize:            o.BatchSize,
 		FlushInterval:        o.FlushInterval,
 		MaxConcurrentUploads: o.BatchMaxConcurrentUploads,
 		MaxInFlightFlushes:   o.BatchMaxInFlightFlushes,
 		AckProgressInterval:  o.BatchAckProgressInterval,
+		DocState:             indexer.NewNATSDocStateStore(docStateKV),
 	}
 
 	// Create separate batchers for audit events and re-indexing events

docs/components/resource-indexer.md

@@ -245,7 +245,41 @@ the buffer reaches its size or time threshold:
 
 Meilisearch applies an index's tasks in the order they are enqueued, so each
 batch enqueues its tasks only after the batch taken before it has enqueued.
-This ordering holds within one indexer replica only.
+That holds within one replica only; across replicas, which share one JetStream
+queue group, the events for one object are spread out and an older event's
+batch can be enqueued after a newer one's.
+
+A document state bucket (JetStream KV, created by the indexer) is the arbiter
+across replicas. It records, per index and document, the newest state any
+replica has accepted — its `resourceVersion`, whether it is indexed, removed
+from the index or deleted, and a copy of the document — plus the latest
+Meilisearch task known to carry a state for it. Reading the current document
+from Meilisearch cannot play that role: writes are queued tasks that take
+longer to apply than a flush interval under load, so a read misses exactly the
+newer write it needs to see. Every flush:
+
+1. Claims each operation with a compare-and-set. An operation older than the
+   recorded state is replaced by that state, so a late create re-applies the
+   newest update instead of the create response.
+2. Enqueues the claimed states and learns the task uid that carries them.
+3. Records that task. If the bucket already names a newer state, it was
+   enqueued before this task and is about to be overwritten, so the flush
+   enqueues it again behind its own task. Only the flush holding the latest
+   recorded task repairs; a flush that fails between steps is nacked and the
+   redelivery runs the protocol from the newest recorded state.
+
+Within one buffer, an operation older than the one already buffered for its
+document is dropped rather than replacing it.
+
+A deletion of the object itself (a `delete` verb, or a write with a
+`deletionTimestamp`) is recorded as a tombstone that wins over every later
+flushed write regardless of `resourceVersion`, since an object's uid is never
+reused. Removal because the object stopped matching a policy is an ordinary
+versioned state, so a newer matching write brings the document back.
+
+Dry-run writes are audited like real ones but persist nothing; the indexer
+skips any event whose request URI carries `dryRun`, and any write whose
+response lacks a `resourceVersion`.
 
 Create events that don't match any policy can be acknowledged immediately since
 the resource was never indexed. Update and delete events that don't match should