Consumer State Reconciliation After Stream Data Loss (R1 Consumers)

[tilakraj94]

Proposed change

Add reconciliation logic for R1 (single replica) consumers when the stream's stored data has reverted due to data loss (e.g., VM crash, corrupted index.db, truncated .blk files). This prevents consumers from being in an inconsistent state where their sequence numbers are ahead of the stream.

stream.zip

This zip file contains stream and consumer data where the stream has no block files and the index file is also empty, resulting in stream sequences being reset to 0, 0. However, the consumer still retains state data, with its sequence at 1844.

When the server is restarted, NATS recovers with the correct stream sequences, but since the consumer sequence is ahead of the stream sequence, messages continue to accumulate instead of being consumed.

Use case

Consumer state should always be consistent with the actual stream state.

When a stream loses data (VM crash, corrupted storage, truncated files), the consumer's stored sequence numbers can become ahead of the stream's actual last sequence. This causes the consumer to become stuck - it won't deliver any messages until the stream accumulates enough new messages to "catch up" to the consumer's sequence.

Example

• Stream has 1000 messages (lseq=1000)
• Consumer has delivered and acked all 1000 (sseq=1001, asflr=1000)
• VM crash occurs - stream data is lost/corrupted
• On recovery: stream has lseq=0 (could be 0 or less than actual consumer seq), but consumer still has sseq=1001, Consumer is now stuck - it's waiting for message 1001, but stream is empty Even if you publish 100 new messages (lseq=100), consumer won't see them because it's waiting for seq 1001. You'd need to publish 1001 messages before the consumer starts working again

Consumers should always reconcile themselves with stream states to make sure consumers have valid sequences. Source of truth for consumers should be stream seq as well.

PS: Sync is default

Contribution

Yes.

[tilakraj94]

As @MauriceVanVeen mentioned in the above PR, regarding the cases, I have tested this method locally.

func (o *consumer) reconcileStateWithStream(streamLastSeq uint64) {
	o.srv.Noticef("Consumer %q reconciling state with stream (lseq=%d): before sseq=%d dseq=%d asflr=%d adflr=%d pending=%d rdc=%d",
		o.name, streamLastSeq, o.sseq, o.dseq, o.asflr, o.adflr, len(o.pending), len(o.rdc))

	// If an ack floor is higher than stream last sequence, reset everything
	if o.asflr > streamLastSeq {
		o.srv.Warnf("Consumer %q ack floor (%d) > stream last seq (%d), resetting all state", o.name, o.asflr, streamLastSeq)

		o.sseq = streamLastSeq + 1
		o.dseq = 1
		o.adflr = 0
		o.asflr = streamLastSeq
		o.pending = nil
		o.rdc = nil
		return
	}

	// Remove pending entries that are beyond the stream's last sequence
	if len(o.pending) > 0 {
		for seq := range o.pending {
			if seq > streamLastSeq {
				delete(o.pending, seq)
			}
		}
	}

	// Remove redelivered entries that are beyond the stream's last sequence
	if len(o.rdc) > 0 {
		for seq := range o.rdc {
			if seq > streamLastSeq {
				delete(o.rdc, seq)
			}
		}
	}

	// Update starting sequence and delivery sequence based on pending state
	if len(o.pending) == 0 {
		o.sseq = o.asflr + 1
		o.dseq = o.adflr + 1
	} else {
		// Find highest stream sequence in pending
		var maxStreamSeq uint64
		var maxConsumerSeq uint64

		for streamSeq, p := range o.pending {
			if streamSeq > maxStreamSeq {
				maxStreamSeq = streamSeq
			}
			if p.Sequence > maxConsumerSeq {
				maxConsumerSeq = p.Sequence
			}
		}

		// Set next sequences based on highest pending
		o.sseq = maxStreamSeq + 1
		o.dseq = maxConsumerSeq + 1
	}

	// Ensure we don't go beyond the stream
	if o.sseq > streamLastSeq+1 {
		o.sseq = streamLastSeq + 1
	}

	o.srv.Warnf("Consumer %q reconciled state: after sseq=%d dseq=%d asflr=%d adflr=%d pending=%d rdc=%d",
		o.name, o.sseq, o.dseq, o.asflr, o.adflr, len(o.pending), len(o.rdc))
}

Few followup questions:

Is introducing a stateLoaded flag the right approach to handle the “valid zero” state, or should we instead adjust the existing condition? Currently, when a consumer is reset to 0,0, the stateWithCopyLocked method checks for both values being zero and triggers a read from disk, which can result in loading stale state.

[MauriceVanVeen]

Have been thinking about this the last couple days, and think it's okay to do this type of consumer reconciliation for R1 consumers in general.

I've reviewed and added comments to: #7692, as well as written some tests that you could cherry-pick into your branch: 0520599

[MauriceVanVeen]

I'm also leaning toward including this in the 2.14 release due in March, instead of including it in a patch version earlier btw.

[tilakraj94]

I'm also leaning toward including this in the 2.14 release due in March, instead of including it in a patch version earlier btw.

That sounds great. Thanks again for all the guidance, I really appreciate it.