Change Feed Processor: Adds Lease container export support

[yash2710]

Description

This PR adds the ability to persist and restore in-memory ChangeFeed processor lease state via a MemoryStream. When a processor is built with WithInMemoryLeaseContainer(MemoryStream), the stream serves as both input and output: existing data in the stream initializes the lease container on startup, and the current lease state is automatically written back to the stream when the processor stops. This enables backup, restore, and restart scenarios for in-memory lease containers without requiring a Cosmos DB lease container.

API

public class ChangeFeedProcessorBuilder
{
    /// <summary>
    /// Uses an in-memory container to maintain state of the leases, optionally
    /// initialized from a MemoryStream containing previously persisted lease state.
    /// When the processor stops, the current lease state is automatically written
    /// back to the same stream.
    /// </summary>
    /// <param name="leaseState">
    /// A MemoryStream that serves as both input and output for lease state. The stream
    /// must be writable and expandable (e.g., new MemoryStream()). A fixed-size stream
    /// such as new MemoryStream(byte[]) will fail at shutdown if the serialized lease
    /// state exceeds the original buffer capacity. StopAsync must not be invoked
    /// concurrently from multiple threads.
    /// </param>
    public virtual ChangeFeedProcessorBuilder WithInMemoryLeaseContainer(
        MemoryStream leaseState);
}

Design Decisions

Single MemoryStream for Read and Write

• The MemoryStream serves dual purpose: if it contains data on init, leases are deserialized from it. On StopAsync, current lease state is serialized back into the same stream.
• This keeps the API surface minimal — one new method, no new types, no separate export/import calls.

Persist-First Shutdown Ordering

• ChangeFeedProcessorCore.StopAsync calls storeManager.ShutdownAsync() before partitionManager.StopAsync().
• If persistence fails, the exception surfaces immediately and partition shutdown is skipped (caller decides what to do). If persistence succeeds, partitions are stopped with the snapshot already durable.
• Trade-off: checkpoint mutations produced during partition shutdown are not captured in the persisted snapshot. Acceptable because change-feed consumers are already required to be idempotent.

Stream Writer Correctness

• DocumentServiceLeaseContainerInMemory.ShutdownAsync calls SetLength(serializedBytes.Length) before writing. If the supplied stream is not expandable and cannot hold the new payload, SetLength throws and the user's stream is left untouched — no partial-write corruption.
• A failed resize surfaces as InvalidOperationException with a message pointing callers at new MemoryStream() instead of new MemoryStream(byte[]).

Automatic Persistence on Stop

• Lease state is written to the stream inside the StopAsync flow, via a virtual ShutdownAsync() lifecycle hook on the internal DocumentServiceLeaseContainer base class, overridden only by the in-memory implementation.

Shared Serialization Format

• Read and write paths both go through InMemoryLeaseJsonFormat, a single internal helper that owns encoding, buffer size, and JsonSerializer settings. This prevents silent drift between writer and reader.

Duplicate Lease Detection

• DocumentServiceLeaseStoreManagerInMemory.DeserializeLeaseState fails fast with InvalidOperationException if the persisted state contains duplicate lease ids, rather than silently overwriting entries.

No Changes to Cosmos-Backed Leases

• The Cosmos-backed lease container is unchanged. The ShutdownAsync base class method is a no-op for implementations that manage their own persistence.

In-Memory Only

• This feature is scoped to in-memory lease containers. Cosmos-backed containers already persist leases in Cosmos DB and don't need this mechanism.

Usage Example

// First run — start with an empty, expandable stream
MemoryStream leaseState = new MemoryStream();

ChangeFeedProcessor processor = container
    .GetChangeFeedProcessorBuilder<MyDocument>("myProcessor", HandleChangesAsync)
    .WithInstanceName("instance-1")
    .WithInMemoryLeaseContainer(leaseState)
    .Build();

await processor.StartAsync();
// ... process changes ...
await processor.StopAsync();
// leaseState now contains the serialized lease state

// Save to file for later use
File.WriteAllBytes("leases.json", leaseState.ToArray());

// Later — restore from saved state.
// IMPORTANT: use an EXPANDABLE MemoryStream. `new MemoryStream(byte[])` is
// non-resizable and will throw at StopAsync if new state is larger than the
// original buffer. Always rehydrate by writing the bytes into a fresh,
// growable MemoryStream and resetting Position to 0.
byte[] savedState = File.ReadAllBytes("leases.json");
MemoryStream restoredState = new MemoryStream();
restoredState.Write(savedState, 0, savedState.Length);
restoredState.Position = 0;

ChangeFeedProcessor newProcessor = container
    .GetChangeFeedProcessorBuilder<MyDocument>("myProcessor", HandleChangesAsync)
    .WithInstanceName("instance-1")
    .WithInMemoryLeaseContainer(restoredState)
    .Build();

await newProcessor.StartAsync();
// Resumes processing from where it left off

Why not new MemoryStream(savedState)? That constructor returns a non-resizable stream backed by the supplied buffer. If the current lease state serializes to even one byte more than savedState.Length, StopAsync will fail — and because nothing fails at startup, the problem only shows up at shutdown in production. The Write + Position = 0 pattern above produces a resizable copy and avoids this class of bug entirely.

Test Coverage

Builder Tests (ChangeFeedProcessorBuilderTests)

• WithInMemoryLeaseContainerWithStreamInitializesStoreCorrectly — Verifies leases are restored from a populated stream.
• WithInMemoryLeaseContainerWithEmptyStreamInitializesEmptyStore — Empty stream creates an empty container.
• WithInMemoryLeaseContainerWithEmptyArrayStreamInitializesEmptyStore — Covers the empty-array seed variant.
• WithInMemoryLeaseContainerWithNullStreamThrows — Validates null argument handling.
• WithInMemoryLeaseContainerWithStreamCannotCombineWithLeaseContainer — Prevents combining with Cosmos container.
• WithInMemoryLeaseContainerWithStreamCannotCombineWithExistingInMemory — Prevents double in-memory configuration.
• WithInMemoryLeaseContainerWithCorruptedStreamThrowsInvalidOperation — Malformed JSON surfaces as InvalidOperationException.
• WithInMemoryLeaseContainer_FullLifecycle_RestoreProcessStopPersist — End-to-end restore → use → stop → re-persist.

In-Memory Container Tests (DocumentServiceLeaseContainerInMemoryTests)

• ShutdownAsync_WithNoStream_IsNoOp — No-op when no stream is configured.
• ShutdownAsync_WritesExpectedCount — Correct lease count serialized (parameterized: 0, 2).
• ShutdownAsync_StreamPositionResetToZero — Stream position reset for consumers.
• ShutdownAsync_WithNonEpkLease_StillSerializes — Non-EPK lease types persist correctly.
• ShutdownAsync_WithDisposedStream_Throws — Disposed stream surfaces as InvalidOperationException.
• ShutdownAsync_WithNonResizableStream_SameSizeData_WritesSuccessfully — Fixed buffer sized exactly right still works.
• ShutdownAsync_WithNonResizableStream_LargerData_ThrowsInvalidOperation — Fixed buffer too small fails fast, before any partial write.
• PersistThenDeserialize_RoundTrip_PreservesData — Full round-trip preserves LeaseToken, ContinuationToken, Owner, Properties (including unicode), FeedRange, and Timestamp.
• PersistOverwritesPreviousStreamContent — Second persist replaces previous data (no stale trailing bytes).
• Deserialize_DuplicateIds_Throws — Duplicate lease ids in persisted state fail fast.
• Deserialize_LeavesStreamPositionAtZero — Reader resets stream position so subsequent writers see a fresh stream.

Processor Core Tests (ChangeFeedProcessorCoreTests)

• StopAsync_CallsShutdownAsync — Verifies ShutdownAsync is invoked during stop.
• StopAsync_WithInMemoryLeases_PersistsStateToStream — Persist-first ordering produces a populated stream.
• StopAsync_WhenShutdownAsyncThrows_ExceptionPropagates — Persistence failure surfaces to the caller and skips partition shutdown.

Type of change

• New feature (non-breaking change which adds functionality)

Closing issues

closes #5580

[github-actions]

All good!

[kushagraThapar]

I am curious if there are multiple version of leases present and if a customer tries to import v1 leases into a lease container where v2 leases are present? Is that a case in .NET SDK?
Also, what happens if existing leases are preserved and imported leases also have the same partitions as preserved leases, which ones take precedence?

[yash2710]

Also, what happens if existing leases are preserved and imported leases also have the same partitions as preserved leases, which ones take precedence?

If there are existing leases, and the overwriteExisting flag is set to true, it would overwrite the preserved leases. The default value for overwriteExisting is false

[kushagraThapar]

Also, what happens if existing leases are preserved and imported leases also have the same partitions as preserved leases, which ones take precedence?

If there are existing leases, and the overwriteExisting flag is set to true, it would overwrite the preserved leases. The default value for overwriteExisting is false

makes sense, I am curious about the case when overwriteExisting is false and existing leases have the partition which importing leases also have, is there conflict resolution happening to decide which lease to take for the specific partition?

[yash2710]

Also, what happens if existing leases are preserved and imported leases also have the same partitions as preserved leases, which ones take precedence?

If there are existing leases, and the overwriteExisting flag is set to true, it would overwrite the preserved leases. The default value for overwriteExisting is false

makes sense, I am curious about the case when overwriteExisting is false and existing leases have the partition which importing leases also have, is there conflict resolution happening to decide which lease to take for the specific partition?

No there is no conflict resolution, it will just skip that lease

[kirankumarkolli]

Please check my comments

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[kirankumarkolli]

@sdkReviewAgent-2

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[kirankumarkolli]

@sdkReviewAgent

[NaluTripician]

PR Deep Review Summary

Thanks for iterating on this — a lot of prior feedback has already been addressed (stream disposal via leaveOpen, ser/deser co-location into the in-memory manager, ShutdownAsync lifted onto DocumentServiceLeaseStoreManager, logging order in StopAsync).

Posting 7 new findings below (1 🔴 Blocking, 4 🟡 Recommendations, 2 🟢/💬). I reviewed all existing comments and skipped items that are already covered by prior reviewers.

Top concern: the runtime guard intended to reject non-resizable MemoryStream inputs (leaseState.SetLength(leaseState.Length)) is ineffective — MemoryStream.SetLength only throws when value > capacity, and for new MemoryStream(byte[]) the length equals capacity at construction. I verified this empirically on .NET 8. Consequence: the documented restore pattern (new MemoryStream(File.ReadAllBytes(...))) passes validation but fails at shutdown the moment serialized state grows, and one of the new unit tests asserts on behavior that does not occur.

Details inline.

---

_{⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.}

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[NaluTripician]

Re-review after commit 40551c6

Thanks for the quick turnaround. Status of my prior findings:

Addressed — thank you:

• 🔴 F1 (ineffective SetLength(Length) guard) — resolved by removing the eager validation entirely.
• 🟡 F4 (WithInMemoryLeaseContainerWithNonResizableStreamThrows was a false-positive test) — removed alongside the guard.
• 🟢 F7 (validation order) — ValidateNoLeaseContainerConfigured() now runs before the argument checks, so the user stream isn't mutated before we decide to reject the call.

Still open from the earlier round (code unchanged; my inline comments stand):

• 🟡 F3 — DocumentServiceLeaseContainerInMemory.ShutdownAsync writes to the stream and then calls SetLength(Position); if the caller passed a stream whose buffer already contained more bytes (e.g. reused from a prior run), the trailing bytes are truncated but can also be partially overwritten in place, leaving a corrupted payload if the write throws mid-flight.
• 🟡 F5 — ChangeFeedProcessorBuilderTests.FullLifecycle constructs the lease-store manager directly and never runs it through the processor, so the test doesn't actually exercise the lifecycle it claims to.
• 🟢 F6 — DeserializeLeaseState silently overwrites duplicate lease IDs instead of failing fast.
• 💬 F9 — DocumentServiceLeaseConverter mutates a shared ContractResolver on a shared JsonSerializer — racy under concurrent use.

---

🟡 New finding from the fix itself · Documentation: Usage Example contradicts the (now-only) guard rail

With the eager validation removed, the XML doc warning is the only defense against non-resizable streams:

"The stream must be writable and expandable (for example, created via new MemoryStream()). A fixed-size stream such as new MemoryStream(byte[]) will fail at shutdown if the serialized lease state exceeds the original buffer capacity."

But the PR description's canonical Usage Example still does the opposite:

byte[] savedState = File.ReadAllBytes("leases.json");
MemoryStream restoredState = new MemoryStream(savedState);  // <-- non-resizable

This example is the thing that gets copied into customer samples, blog posts, and StackOverflow answers. Users who follow it will:

1. Restore from a byte array → stream is non-resizable.
2. Start the processor → works fine (only reads during init).
3. Acquire new leases / continuation tokens grow / owner changes → serialized state grows even a little.
4. StopAsync throws InvalidOperationException (Memory stream is not expandable.) at shutdown — exactly the worst time, and with no prior signal.

Because the runtime guard has been removed, the documented pattern now silently succeeds until it bites in production.

Suggested fix — update the Usage Example (both in the PR description and anywhere this lands in customer-facing docs / <example> XML) to explicitly produce a resizable stream:

byte[] savedState = File.ReadAllBytes("leases.json");
MemoryStream restoredState = new MemoryStream();
restoredState.Write(savedState, 0, savedState.Length);
restoredState.Position = 0;

Or wrap as a helper and call it out in a comment: // Resizable copy — new MemoryStream(byte[]) would fail at shutdown.

---

Also still unresolved from @xinlian12's earlier comment: if partitionManager.StopAsync() throws inside ChangeFeedProcessorCore.StopAsync, ShutdownAsync on the lease-store manager is skipped, so the in-memory lease state never gets persisted to the caller's stream. Worth deciding whether that path should use a try/finally.

---

_{⚠️ AI-generated review — may be incorrect. Agree? → resolve the conversation. Disagree? → reply with your reasoning.}

[xinlian12]

✅ Review complete (44:03)

No new comments — existing review coverage is sufficient.

_{Steps: ✓ context, correctness, cross-sdk, design, history, past-prs, synthesis, test-coverage}

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).

[azure-pipelines]

Azure Pipelines:
Successfully started running 1 pipeline(s).