Egil.Orleans.Messaging#58
Conversation
|
Review findings from the feature branch review:
Signed-off-by: GPT-5 |
|
@copilot resolve the merge conflicts in this pull request |
Conflicts resolved in |
egil
changed the title
Comprehensive API design document covering all 8 design sections: - §0: Scope, packaging (one NuGet, split-ready), name settled as Egil.Orleans.Messaging - §1: IStateManager<T> atomic writes with in-flight recovery - §2: Outbox<T> storage placement (co-located on grain state) - §3: Outbox<T> sealed class shape, epoch semantics, fingerprint equality, configurable max depth - §3a: VersionedState<TSelf> for ImmutableArray equality trap - §4: MessageTracker sealed class, dual-dictionary dedup - §5: StreamManager fluent builder facade - §6: Opinionated functional-grain pattern (guidance, not enforced) - §7: OutboxProcessor<T> grain-scoped timer+reminder dispatch with callback-based postmen, per-attempt error tracking - §8: Naming (flat namespace), serialization ([Alias] with fully qualified prefix, sequential [Id]), telemetry conventions All decisions settled via structured grilling sessions. Directory still named Egil.Orleans.CQRS — rename deferred to project scaffold. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Rename directory from Egil.Orleans.CQRS to Egil.Orleans.Messaging. Add solution file, Directory.Packages.props, version.json, global.json, src csproj (library), and test csproj. Package description: toolbox of composable building blocks — not a framework. Pick what you need, leave the rest. [skip notes] Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Limit StreamCursorJsonConverter to Orleans built-in stream sequence token types and remove custom token adapter serialization paths. Unsupported token kinds now throw a NotSupportedException with an actionable message that explains the stability rationale and points users to the GitHub issues page for feature requests. Update StreamCursorJsonConverterTests to assert unsupported custom tokens and unknown token kinds fail with the new guidance message while keeping built-in token round-trip coverage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Enable JSON converter attribute on OutboxSequenceToken and implement OutboxSequenceTokenJsonConverter read/write paths. The converter now round-trips sequence, sender GrainId shape, timestamp, and epoch without requiring caller-side serializer options. Add focused tests for converter attribute registration, round-trip behavior (including multiple GrainId key shapes), and payload validation for missing sender. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Enable JSON converter factory attribute on OutboxMessageEnvelope<T> and implement OutboxMessageEnvelopeJsonConverterFactory with closed generic converters for read/write round-trip serialization. Add tests for attribute wiring, CanConvert behavior, round-trip serialization, and missing-token payload validation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Implement StateManager<T> state cache, read/write/clear operations, force-write ETag override, and conflict handling that reconciles equivalent persisted state before rethrowing non-equivalent failures. Add focused tests for read refresh, successful writes, write-failure reconciliation behavior, rollback on conflicting writes, rollback when post-failure read fails, and VersionedState version stamping. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Implement AsStateManager<T> to validate input and wrap IPersistentState<T> with StateManager<T> for grain-facing usage. Add focused tests for successful wrapping and null-argument validation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Enable StreamCursor JSON converter attribute and implement TryGetEnqueuedTime, TryGetStreamProviderName, and TryGetTraceParent to expose enrichment metadata when backed by EnrichedEventHubSequenceToken. Add focused tests for positive and negative cases across enqueued time, stream provider name, and traceparent access. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Implement UseEnrichedDataAdapter extension to validate input and register EnrichedEventHubAdapter through the Event Hub configurator using Orleans Serializer from DI. Add focused tests for null-argument validation and configurator callback registration. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Implement EnrichedEventHubAdapter overrides to stamp traceparent on queued messages, create EnrichedEventHubSequenceToken from cached Event Hub messages, and return enriched stream positions that preserve offset/sequence/event index and optional traceparent. Add focused tests for traceparent stamping behavior and enriched token construction from cached message metadata. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Add InitializeStateManager(...) that resolves keyed IStateManagerFactory<T> by storage name, with IServiceCollection and ISiloBuilder helper APIs for keyed singleton registrations. Refactor state management to StateManagerBase<T>/DefaultStateManager<T>, remove WritePolicy and force-write behavior, and drop reflection-based ETag mutation. Update state manager tests and API design docs to match the new factory wiring and write semantics. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
StreamManager now separates subscription configuration from activation-time subscription setup via RegisterStreamManager, AddSubscription, and SubscribeAsync. This lets grains await Orleans stream subscription establishment without blocking activation and keeps subscription failures visible. MessageTracker now records receive-lag histograms for accepted enriched stream cursors and outbox tokens, with shared telemetry plumbing kept in MessagingTelemetry. The Messaging test suite adds an in-process Orleans cluster with memory streams covering ValueTask, Task, default error handling, and resume-token behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Rename InitializeStateManager to RegisterStateManager so state and stream activation helpers use the same registration language before the package ships. Update state manager docs and tests to match the new API name. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Capture the current OutboxProcessor design direction, including background posting, timer interleaving options, postman execution modes, and related telemetry guidance. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Fix stream subscription initialization so handles are collected without concurrent List mutation, repeated SubscribeAsync calls are rejected, and stream handlers receive nullable sequence tokens instead of relying on null-forgiving calls. Recover StateManager clear failures with a read-back path, clamp negative receive-lag telemetry, validate malformed outbox JSON with JsonException, and document the intentional outbox equality fingerprint invariant. Rename the deferred outbox processor activation API references to RegisterOutboxProcessor to align with the other activation-time helpers while leaving the implementation deferred. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Implements the outbox processor registration and draining path, including postman dispatch, success acknowledgement, failure reconciliation, retry timer/reminder scheduling, and stream-backed integration coverage. Adds outbox and stream receive telemetry plus focused Orleans in-process tests for stream delivery, error paths, resume behavior, JSON converter branches, and tracker semantics. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Adds targeted tests for high-risk converter and tracker branches, including stream cursor token variants, malformed persisted JSON, and same-count MessageTracker equality mismatches. These tests raise branch coverage around runtime serialization and dedup semantics without padding low-value argument validation paths. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Stream cursors now carry namespace and provider metadata instead of using the Orleans stream id as the persisted tracking identity. Provider names are populated from subscription handles, with enriched Event Hub tokens taking precedence when present. This keeps tracking grain-local while still supporting provider-aware deduplication for non-conventional multi-provider namespace usage. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Document the Messaging package entry points, add it to the repository package list, and register its conventional commit scope. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Add the Messaging package workflow, include the package README in the NuGet artifact, and make Messaging tests consume Egil.Orleans.Testing as a package dependency instead of a sibling project reference. [skip notes] Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Add keyed outbox postman tests for caller cancellation and processor timeout, verifying both paths leave the outbox item pending and do not acknowledge or reconcile it as a per-item failure. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Add built-in stream and grain postman helpers so common Orleans fan-out targets can use the outbox processor without hand-written dispatch callbacks. Cover helper behavior through Orleans tests and fill validation coverage for keyed postman registration edge cases. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Fail fast when Orleans invokes an implicit stream subscription without a configured handler, and add explicit StreamId subscription overloads so explicit streams are not limited to the grain-keyed convention. Clarify outbox ownership, first-match postman dispatch, thread-pool postman safety, and future provider-specific package boundaries in public docs. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Move Event Hubs stream enrichment into Egil.Orleans.Messaging.Streams.EventHubs so the core package no longer carries the Event Hubs dependency. Add Egil.Orleans.Messaging.State.AzureStorage with Azure Table/Blob storage-aware state manager registration and failure classification for optimistic concurrency failures. Rename WriteFailureKind to StorageFailureKind so write and clear recovery can share the same provider classification model. Update CI to pack all messaging packages and document the companion package model. BREAKING CHANGE: Event Hubs enrichment types moved out of Egil.Orleans.Messaging into Egil.Orleans.Messaging.Streams.EventHubs, and WriteFailureKind was renamed to StorageFailureKind. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Classify deterministic Azure Storage failures as non-persistence outcomes and keep ambiguous transient results on the read-back recovery path. This lets StateManager skip unnecessary recovery reads for rejected mutations while preserving correctness for throttling, timeouts, and service failures. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Dispatch pending outbox items concurrently on the Orleans activation scheduler and remove the thread-pool execution mode. Background posting now allows delivery to interleave by default while durable acknowledgement and failure reconciliation remain non-interleaving unless explicitly opted in. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Add an explicit JsonConverter registry for concrete StreamSequenceToken types and route StreamCursor and MessageTracker token payloads through that registry. Register Event Hubs token converters from the enriched adapter setup so enriched checkpoints round-trip without downcasting provider metadata. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Document the Azure Storage classification behavior, outbox background posting model, and explicit stream token JSON converter registry. Keep the public README focused on observable behavior while leaving Orleans scheduling rationale in the API design notes. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Use a non-generic IStateManagerFactory with a generic Create<T> method so keyed registrations are easier to reason about and provider-specific factories do not need open generic registrations. Also simplifies Azure Storage failure code matching and documents the ambiguous clear recovery path.
Remove posted outbox tokens by identity so successful items after a failed gap can still be acknowledged. Also makes outbox token/envelope types Orleans-friendly with required init properties and rewrites the STJ converters to delegate through typed converters while preserving the existing JSON shape.
Do not persist a stream tracking entry when Orleans delivers a null StreamSequenceToken. Without a token there is no high-water mark to compare, so tokenless stream messages are accepted without changing the tracker instead of causing later tokenless messages to look like duplicates.
Refactor outbox processing into dedicated dispatch, postman registry, and reconciliation collaborators. Coordinate manual and timer-backed drains through a single active drain so callers wait for the current pass before processing the next pending snapshot. Preserve per-postman ordering while allowing different postmen to dispatch concurrently. Remove the prototype OutboxProcessor2 implementation from the compiled package.
A failed PostAsync run - processing timeout, cancellation, or an exception from an acknowledgement or reconciliation callback - previously skipped the retry scheduling that normally happens after reconciliation. On a first-ever foreground post no timer or reminder existed yet, so announced messages stayed in the durable outbox until the next post or activation. The processor now arms its retry timer and reminder before rethrowing the post-run failure, but only on the failure path, so successful posts still avoid the reminder registration write. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Outbox equality now compares the full first and last pending tokens, including their timestamps, instead of only their sequence numbers. Items are only appended at the tail, so two outbox histories that diverged by adding different messages - for example duplicate grain activations racing an ambiguous storage write - always differ in their highest pending token. Comparing the full token lets the state-manager recovery read-back distinguish such diverged states instead of mistaking them for a lost response, which could silently drop the local activation's message. Histories that diverged only by removals can still compare equal; recovery then at worst re-delivers an already posted item (at-least-once) but never loses a pending message. Equality remains O(1) and still ignores payloads. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
PostInBackgroundAsync and follow-up drain scheduling no longer register the durable reminder up front; they arm only the in-memory grain timer. The reminder - which costs a storage write - is now registered only when a post run fails or completes with items still pending, including when a background dispatch or reconciliation callback throws. Unregistering also caches the one-time lookup for a leftover reminder from a previous activation, so an empty outbox no longer pays a reminder-table read on every successful drain. Successful posts therefore incur no reminder I/O. The trade-off is that a silo crash between scheduling a background post and the first failed run loses the cross-activation retry trigger; pending items then wait for the next activation's post instead of a reminder tick. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Expand the Outbox<T>.Equals XML docs to fully explain the state-manager write-recovery contract the fingerprint equality must satisfy: why a false positive would drop pending messages, why the append-at-tail invariant plus full first/last tokens (incl. activation timestamps) keep the O(1) comparison loss-safe, that ETag conflicts are rethrown separately and equality only decides ambiguous-outcome writes, and the accepted at-least-once trade-off for removal-only divergence. Also document the same contract at the consumer side on StateManagerBase.IsEquivalent and note on FirstToken/LastToken why full tokens are compared instead of bare sequence numbers. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
The outbox processor tracks per-item dispatch attempt counts in an in-memory dictionary keyed by item equality. Entries were only removed when an item posted successfully, so items that left the outbox any other way - dead-lettered or dropped in ReconcileFailedAsync, or removed directly by the grain - kept their entries for the lifetime of the activation, a slow memory leak invisible to the owning grain. The processor now prunes counters for items no longer pending after each reconciliation. A consequence is that an equal item enqueued after its predecessor was dead-lettered starts a fresh attempt sequence instead of inheriting stale counts. Also documents the attempt-count semantics on ReconcileFailedAsync: counts are per-activation and keyed by item equality, so retry and dead-letter policies that must survive activation restarts need to persist their own counters. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
The library is a toolbox: grains may use Outbox and OutboxProcessor without the state manager. Document what those users can rely on and where the sharp edges are so problems stay detectable: - README: the outbox pipeline is at-least-once on its own - items only leave durable state via AcknowledgePostedAsync after a successful post, so failed or ambiguous state writes cause duplicate delivery, never loss. Also notes that attempt counts are per-activation and that Outbox<T>.Equals is a fingerprint, not payload equality. - Outbox<T> class docs: equality is exact within a single history lineage (every mutation changes the fingerprint, so dirty-checks are safe) and can only compare equal across divergent activation histories that differ by removals. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
OutboxMessageEnvelope<T> carried SendAttempts and LastSendError members that nothing in the library read or wrote. They contradicted the envelope's documented design - retry diagnostics are tracked in-memory by the outbox processor, not persisted on the envelope - and SendAttempts was needlessly serialized with grain state. Removed before the first stable release. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…lan doc [skip notes] Adds XML docs to the StateManagerBase<T> and DefaultStateManager<T> constructors, the last undocumented public members in the shipped packages, and removes outbox-postman-api-plan.md now that the keyed postman API it planned is implemented and described in api-design.md and the README. The plan remains available in git history. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Adds a src-level Directory.Build.props that re-enables CS1591 (missing XML comment on publicly visible member), which the repo-wide props file suppresses. All public API in the three shipped packages is documented, and Release builds treat warnings as errors, so undocumented public members now fail CI for the src projects while tests keep the suppression. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
The README outbox example snapshotted only message payloads and acknowledged with Outbox.Take(items.Length), assuming posted items form a FIFO prefix. They do not: postmen dispatch their groups concurrently, and an item without a matching postman fails in place while later items succeed. Copying the example into such a setup removes a failed, undelivered item from the outbox and loses it. The example now uses OutboxMessageEnvelope<T> as the processor item type and removes exactly the posted items by token. The AcknowledgePostedAsync contract documents that the batch is not necessarily a contiguous prefix and that acknowledgement must match items, never positions. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
MessageTracker.LatestOutbox reconstructed the last accepted OutboxSequenceToken using the receiver's Received wall-clock time as the token Timestamp. Token equality includes Timestamp, so whenever sender and receiver clocks differ the returned token did not equal the token that was actually accepted, breaking equality-based consumers and any logic reading the timestamp as sender time. Outbox entries now store the sender-stamped timestamp of the last accepted token alongside the receiver-side Received time (which remains the basis for eviction). LatestOutbox returns the original timestamp. JSON payloads written before the new property fall back to Received, matching the previous behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
egil
left a comment
egil
left a comment
There was a problem hiding this comment.
Consolidated release-readiness review (five independent per-area deep reviews, cross-examined; premises verified against Orleans 10.1.0 sources). Build: 0 warnings. Tests: 186/186.
Verdict: not ready for 0.1-alpha yet. Core blocked by 3 P1s (outbox handoff deadlock, Outbox<T> rehydration NRE, StateManager lost update); EventHubs blocked by 1 P1 (enrichment inert — its sole feature). AzureStorage state package is ready (P3s only). All fixes are small and localized — see inline comments for details and suggested fixes.
Not inline-commentable: .github/workflows/prepare-release.yml project list lacks Egil.Orleans.Messaging (noted in the csproj comment). Additional P2 test gaps beyond the inline ones: ISE recovery branches have zero coverage; ClearAsync record-still-exists branch untested; reminder lifecycle never verified end-to-end (fixture already has an in-memory reminder service).
| /// <summary> | ||
| /// Non-persisted service reference. No <c>[Id]</c>, no serialization. | ||
| /// Falls back to <see cref="TimeProvider.System"/> when not explicitly set. | ||
| /// </summary> | ||
| [NonSerialized] | ||
| [JsonIgnore] | ||
| private TimeProvider time = TimeProvider.System; |
There was a problem hiding this comment.
P1 — NRE on first mutation after Orleans-serializer rehydration. This doc claim is false on the paths [GenerateSerializer] exists for: the class has no public parameterless ctor, so Orleans' DefaultActivator materializes instances via GetUninitializedObject — field initializers never run and time == null. Verified against the generated Codec_Outbox/Copier_Outbox (both use _activator.Create()): a binary round-trip and a DeepCopier.Copy (the MemoryGrainStorage read path) both produce an outbox whose next Add throws NRE at time.GetUtcNow() (line 206). Any consumer persisting Outbox<T> with Orleans-serialized grain storage (memory storage in dev, Azure providers on the default OrleansGrainStorageSerializer) crashes on the first Add after reactivation. JSON persistence is unaffected (converters call the real ctor), which is why the test suite misses it — no test mutates a rehydrated outbox.
Fix: read time through a null-tolerant accessor (time ?? TimeProvider.System) at the four use sites, or add a public parameterless ctor like MessageTracker (line 85) has for exactly this reason. A non-public ctor won't help — DefaultActivator only looks up public ones (verified in Orleans 10.1.0 DefaultActivator.Init(): GetConstructor(Type.EmptyTypes)). Also note StateManager recovery adoption (StateManager.cs:153) swaps in a freshly deserialized instance, so even grains that call RegisterTimeProvider at activation lose the registration mid-lifetime; a null-tolerant accessor handles that too.
There was a problem hiding this comment.
Fixed in 2ebf3f4. Outbox<T> now reads its non-serialized clock through a C# 14 field-backed property which falls back to TimeProvider.System after Orleans rehydration, while RegisterTimeProvider still overrides it. Its XML remarks now explicitly require re-registering custom clocks after every deserialization or rehydration when timestamps affect tests or business logic. Added a regression test using Orleans DeepCopier.Copy; it reproduced the NRE before the fix. Validation: 187/187 Release tests passed, and the Release build completed with 0 warnings.
| if (reconciliation.HasWork) | ||
| { | ||
| pendingReconciliation = reconciliation; | ||
| backgroundDrainOwnsActiveDrain = true; | ||
| EnsureReconciliationTimer(TimeSpan.Zero); | ||
| return; |
There was a problem hiding this comment.
P1 — activation deadlock: drain-gate handoff vs non-interleaving reconciliation vs non-reentrant PostAsync. When a dispatch finds work (the common case), this returns holding activeDrain — only ReconcileInBackgroundAsync's finally releases it. The dispatch timer interleaves (default Interleave = true) but the reconciliation timer does not (default InterleaveReconciliationCallbacks = false), so the tick is an ordinary blocking grain request. Sequence on a non-reentrant grain: dispatch-with-work parks the gate → a user call gets scheduled first and awaits PostAsync() → WaitForTurnAsync awaits activeDrain.Task while being the activation's current blocking request → the queued non-interleaving reconciliation tick can never run → CompleteDrain never happens → activation permanently wedged; every subsequent call times out. The window recurs on every dispatch that posts or fails anything, including periodic RetryDelay ticks. The tests prove each leg separately but never the combination — the only concurrent-foreground-PostAsync test grain is [Reentrant]. Also contradicts api-design.md:1226-1230 (PostAsync during an active run should coalesce, not block).
Fix: don't park callers on activeDrain while a background handoff is pending — complete the drain gate before arming the reconciliation tick (give reconciliation its own gate), or make PostAsync coalesce (set drainRequested and return) when backgroundDrainOwnsActiveDrain.
There was a problem hiding this comment.
Fixed in b822c22. A real non-reentrant Orleans cluster test reproduced the cycle: an interleavable background dispatch held activeDrain, a foreground PostAsync occupied the activation waiting for it, and the queued non-interleaving reconciliation turn could not run. The processor now releases the drain gate at the dispatch/reconciliation turn boundary, reconciliation reacquires it, and a foreground drain which wins the race reconciles the pending posted batch before taking a fresh snapshot. Competing background dispatches defer while reconciliation is pending, avoiding duplicate delivery. This preserves direct PostAsync completion semantics instead of returning early. Validation: focused regression, all 16 outbox coverage tests, and the full 188-test Release suite passed.
| if (options.PendingItems().IsDefaultOrEmpty) | ||
| { | ||
| await DisableRetryAsync(); | ||
| return; | ||
| } |
There was a problem hiding this comment.
P2 — empty-path DisableRetryAsync can dispose a pending reconciliation and leak the drain gate. If the outbox becomes empty out-of-band (Clear/RemoveRange in a user call) between the HasWork handoff and the reconciliation tick, a PostInBackgroundAsync/ReceiveReminderAsync arriving in that window takes this empty branch → DisableRetryAsync disposes the armed reconciliation timer (lines 564-565) → the pending tick never fires → ack/reconcile callbacks are lost and activeDrain is never completed — the same permanent wedge as the dispatch-handoff deadlock, plus the already-delivered batch is re-sent after reactivation.
Fix: skip (or defer) DisableRetryAsync while pendingReconciliation != null || backgroundDrainOwnsActiveDrain.
| internal void AttachToGrain() | ||
| { | ||
| owner.GrainContext.SetComponent<IOutboxComponent>(this); | ||
| } |
There was a problem hiding this comment.
P2 — second RegisterOutboxProcessor on one grain silently kills the first outbox's durable retry. SetComponent<IOutboxComponent> is a single slot, but reminder names embed typeof(TOutbox).FullName — the API clearly contemplates multiple outboxes per grain. With two registered: A's items fail → A registers reminder → grain deactivates → reminder fires → the IOutboxGrain DIM forwards to the last-attached component (B) → name mismatch → no-op. A's pending items are never retried cross-activation and A's reminder is never unregistered — it re-activates the grain every period, forever, doing nothing (OnActivateAsync re-registers both processors in the same order, so the same loser keeps losing the slot deterministically). Nothing throws, logs, or documents a one-processor-per-grain limit.
Fix: keep a component list and dispatch by reminder name, or throw on second attach and document the single-outbox constraint.
| var activeReminder = reminder; | ||
| if (activeReminder is null && !checkedForExistingReminder) | ||
| { | ||
| activeReminder = await owner.GetReminder(reminderName); | ||
| checkedForExistingReminder = true; | ||
| } | ||
|
|
||
| if (activeReminder is not null) | ||
| { | ||
| await owner.UnregisterReminder(activeReminder); | ||
| reminder = null; | ||
| } |
There was a problem hiding this comment.
P3 — failed unregister is never retried within the activation. checkedForExistingReminder = true is set before UnregisterReminder, and the looked-up activeReminder is never assigned to the reminder field. If the unregister throws, every later DisableRetryAsync skips both the lookup and the unregister — the stale durable reminder keeps firing each period, re-activating the grain to no-op. Fix: assign reminder = activeReminder before unregistering, or set the flag only after success.
| if (token.Epoch == entry.Epoch && token.SequenceNumber > entry.LastSequenceNumber) | ||
| { | ||
| MessagingTelemetry.RecordOutboxReceiveLag(token, now); | ||
| next = CreateTracker(streams, outbox.SetItem(token.Sender, new OutboxEntry(entry.Epoch, token.SequenceNumber, now, token.Timestamp))); | ||
| return true; | ||
| } |
There was a problem hiding this comment.
P2 (test gap) — the steady-state accept branch is untested; an inverted comparison passes all 186 tests. This same-epoch/newer-sequence branch is the single most-traveled dedup path in production, and the logic is correct (verified: == epoch + > sequence; DateTimeOffset > is instant-based and offset-safe) — but no test pins it. MessageTrackerTests covers accept-first, reject-same-seq, reject-stale-epoch, accept-newer-epoch; every cluster test sends exactly one outbox-token message per grain pair, and the projected sink dedups via stream cursors. Flipping > to < here survives the entire suite. One ~5-line unit test closes it: register (epoch E, seq 1) → (E, seq 2) accepted → (E, seq 2) again rejected.
| if (!reader.ValueTextEquals(nameof(StreamCursor.ProviderName))) | ||
| { | ||
| throw new JsonException($"Expected StreamCursor property '{nameof(StreamCursor.ProviderName)}', got '{reader.GetString()}'."); | ||
| } |
There was a problem hiding this comment.
P3, but decide before v1 payloads exist — strict schema breaks rolling upgrades. Read requires exact property order and throws on any unknown property (also lines 83-86; same envelope pattern in StreamSequenceTokenJsonConverters and the EventHub token converters). Any future added property makes payloads written by new silos unreadable by old silos during a rolling upgrade or rollback — JsonException during grain state load means the grain fails to activate. This strictness ships permanently with v1 readers and can't be retrofitted once payloads exist in production storage. Fix: property-name-dispatch loop that tolerates order and skips unknowns, keeping writes canonical — v1 readers then get forward tolerance for free.
| JsonSerializer.Serialize( | ||
| writer, | ||
| new MessageTrackerJsonModel(streamModels, outboxModels), | ||
| options); |
There was a problem hiding this comment.
P3, but fix before release — wire format is naming-policy sensitive; a mismatch silently wipes all dedup state. The nested private models serialize via the ambient JsonSerializerOptions, so a PropertyNamingPolicy (e.g. web-defaults camelCase) changes the persisted property names — unlike StreamCursorJsonConverter/OutboxSequenceTokenJsonConverter, which hardcode names (inconsistent robustness, and the opposite failure mode of their strictness). Reading PascalCase-persisted state under camelCase options (or vice versa) leaves model.Streams/model.Outboxes null, and the coalesce-to-empty at lines 48/57 makes the tracker deserialize as empty, silently — all dedup state lost, previously rejected duplicates accepted. Fix: [JsonPropertyName] on the model records or explicit property writes like the sibling converters; consider failing loudly when both arrays are absent.
| <PropertyGroup> | ||
| <PackageId>Egil.Orleans.Messaging</PackageId> | ||
| <Title>Orleans Messaging Toolbox</Title> | ||
| <Company>Egil Hansen</Company> | ||
| <Authors>Egil Hansen</Authors> | ||
| <Description> | ||
| A toolbox of composable building blocks for Orleans grains that need | ||
| atomic state writes, transactional outbox, receiver-side dedup, and | ||
| stream subscription management. Pick what you need, leave the rest. | ||
| </Description> | ||
| <PackageReadmeFile>README.md</PackageReadmeFile> | ||
| <PackageTags>orleans,messaging,outbox,deduplication,streams,grain,egil</PackageTags> | ||
| <Copyright>Egil Hansen</Copyright> | ||
| <PackageProjectUrl>https://github.com/egil/framework</PackageProjectUrl> | ||
| <RepositoryUrl>https://github.com/egil/framework</RepositoryUrl> | ||
| <RepositoryType>git</RepositoryType> | ||
| <PackageLicenseFile>LICENSE</PackageLicenseFile> | ||
| <PublishRepositoryUrl>true</PublishRepositoryUrl> | ||
| <EmbedUntrackedSources>true</EmbedUntrackedSources> | ||
| <EnablePackageValidation>true</EnablePackageValidation> | ||
| </PropertyGroup> |
There was a problem hiding this comment.
P2 (packaging) — no release-notes wiring: packages will ship without PackageReleaseNotes. Sibling packages (Egil.Orleans.Testing, Egil.SystemTextJson.Migration) have a scripts/generate-release-notes.ps1 plus a _GenerateReleaseNotes target in the core csproj (<Exec> before GenerateNuspec setting PackageReleaseNotes), and root AGENTS.md documents release notes as generated from commit history per project. This project has neither the script nor the target — same gap applies to the other two csproj files in this PR. Port the script + target from Egil.Orleans.Testing before tagging. Related (not in this diff, flagging here): .github/workflows/prepare-release.yml's project choice list doesn't include Egil.Orleans.Messaging, so a stable release branch can't be prepared via the workflow.
| <PackageReference Include="Azure.Core" /> | ||
| <PackageReference Include="Azure.Data.Tables" /> | ||
| <PackageReference Include="Azure.Storage.Blobs" /> | ||
| <PackageReference Include="Microsoft.Orleans.Persistence.AzureStorage" /> |
There was a problem hiding this comment.
P3 — three of these four references are unused. The compiled source only touches Azure.Core types (RequestFailedException, ETag). Azure.Data.Tables, Azure.Storage.Blobs, and Microsoft.Orleans.Persistence.AzureStorage are never referenced in code — the state manager works with any storage provider that throws RequestFailedException, which is broader than the package name/description suggests. Each unused ref forces unnecessary dependencies (the Orleans one pulls a whole provider) onto consumers who may use only one storage backend, and creates false version coupling. Drop them — or, if the intent is "batteries included" for Azure users, that's worth a deliberate note in the README instead.
There was a problem hiding this comment.
Pull request overview
This PR introduces a new top-level project, Egil.Orleans.Messaging, adding composable messaging infrastructure for Microsoft Orleans grains (durable outbox, receiver-side dedup/tracking, stream cursor/token handling, and resilient state write/clear wrappers), along with provider-specific extensions and CI/versioning scaffolding.
Changes:
- Adds new
Egil.Orleans.Messagingpackages (core + Event Hubs streams + Azure Storage state manager) and supporting telemetry/serialization infrastructure. - Adds comprehensive test projects covering JSON converters, outbox behavior, stream cursor behavior, and state manager recovery semantics.
- Adds project wiring: repo README entry, Conventional Commit scope, per-project versioning files, and a dedicated GitHub Actions workflow.
Reviewed changes
Copilot reviewed 91 out of 92 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| README.md | Adds Orleans Messaging project entry |
| Egil.Orleans.Messaging/version.json | Adds NBGV versioning for project |
| Egil.Orleans.Messaging/global.json | Adds per-project SDK/test runner config |
| Egil.Orleans.Messaging/Egil.Orleans.Messaging.slnx | Adds solution for new project + tests |
| Egil.Orleans.Messaging/Directory.Packages.props | Adds central package versions for project |
| Egil.Orleans.Messaging/src/Directory.Build.props | Re-enables public API XML doc enforcement (CS1591) for this project |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Egil.Orleans.Messaging.csproj | Adds core package project file |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/MessagingTelemetry.cs | Adds metrics + tracing hooks for messaging/outbox/streams |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Tracking/MessageTrackerJsonConverter.cs | Adds STJ converter for MessageTracker persistence |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/IStreamManagerComponent.cs | Adds internal stream manager lifecycle component contract |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/IImplicitStreamGrain.cs | Adds DIM for implicit subscription forwarding |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/IStreamSequenceTokenMetadata.cs | Adds optional metadata contract for enriched tokens |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/StreamCursor.cs | Adds StreamCursor value object (namespace + token + provider name) |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/StreamCursorJsonConverter.cs | Adds STJ converter for StreamCursor |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/StreamManagerExtensions.cs | Adds grain extension to register StreamManager |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/StreamSequenceTokenJsonConverters.cs | Adds process-wide token-kind registry + built-in converters |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Streams/StreamSequenceTokenJsonConverterServiceCollectionExtensions.cs | Adds DI helpers for registering token converters |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/IStateManager.cs | Adds committed-state-fence state manager contract |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/IStateManagerFactory.cs | Adds factory contract for state manager creation |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/StateManager.cs | Adds base implementation with recovery + clear handling |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/DefaultStateManager.cs | Adds default provider-agnostic state manager |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/DefaultStateManagerFactory.cs | Adds default state manager factory |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/VersionedState.cs | Adds version-stamped state base for safe recovery comparisons |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/StorageFailureKind.cs | Adds outcome classification enum for recovery decisions |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/StateManagerExtensions.cs | Adds grain extension to register keyed state managers |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/StateManagerRegistrationExtensions.cs | Adds IServiceCollection keyed factory registrations |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/State/StateManagerSiloBuilderExtensions.cs | Adds ISiloBuilder registration helpers |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/IPostman.cs | Adds outbox delivery contract |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/IOutboxGrain.cs | Adds outbox grain marker + DIM reminder forwarding |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxPostmanAttribute.cs | Adds attribute-based postman naming |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxPostmanServiceCollectionExtensions.cs | Adds keyed DI registration helpers for postmen |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxPostmanRegistry.cs | Adds activation-local postman dispatch registry |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/NoPostmanRegisteredException.cs | Adds explicit error for missing postman mapping |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxDispatchResult.cs | Adds dispatch result record for dispatcher/reconciler |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxReconciliationBatch.cs | Adds reconciliation batch struct (posted/failed) |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxReconciler.cs | Adds attempt-count tracking + reconciliation behavior |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxProcessorOptions.cs | Adds outbox processor configuration object |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxProcessorExtensions.cs | Adds grain extension to register OutboxProcessor |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxDispatcher.cs | Adds concurrent group dispatch with timeout telemetry |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxSequenceToken.cs | Adds outbox token type (sender/seq/time/epoch) |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxSequenceTokenJsonConverter.cs | Adds STJ converter for OutboxSequenceToken |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/GrainIdJsonConverter.cs | Adds internal STJ converter for GrainId payloads |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxMessageEnvelope.cs | Adds envelope type for stored outbox messages |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxMessageEnvelopeJsonConverterFactory.cs | Adds STJ converter factory for OutboxMessageEnvelope |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging/Outboxes/OutboxJsonConverterFactory.cs | Adds STJ converter factory for Outbox |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.Streams.EventHubs/Egil.Orleans.Messaging.Streams.EventHubs.csproj | Adds Event Hubs extension package project |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.Streams.EventHubs/EventHubs/EnrichedEventHubAdapterExtensions.cs | Adds stream configurator extension wiring for enriched adapter |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.Streams.EventHubs/EventHubs/EnrichedEventHubSequenceToken.cs | Adds enriched Event Hubs token carrying metadata + traceparent |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.Streams.EventHubs/EventHubs/EventHubStreamSequenceTokenJsonConverters.cs | Adds Event Hubs token STJ converters + registry registration |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.State.AzureStorage/Egil.Orleans.Messaging.State.AzureStorage.csproj | Adds Azure Storage state-manager package project |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.State.AzureStorage/AzureStorage/AzureStorageStateManager.cs | Adds Azure-specific failure classification + state manager |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.State.AzureStorage/AzureStorage/AzureStorageStateManagerRegistrationExtensions.cs | Adds IServiceCollection registration helper |
| Egil.Orleans.Messaging/src/Egil.Orleans.Messaging.State.AzureStorage/AzureStorage/AzureStorageStateManagerSiloBuilderExtensions.cs | Adds ISiloBuilder registration helper |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Egil.Orleans.Messaging.Tests.csproj | Adds core test project |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/GlobalUsings.cs | Adds shared usings for tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/xunit.runner.json | Adds xUnit runner settings |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/MessagingTestClusterFixture.cs | Adds Orleans test cluster fixture for integration-style tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Tracking/MessageTrackerJsonConverterTests.cs | Adds MessageTracker STJ converter tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Streams/StreamCursorTests.cs | Adds StreamCursor metadata extraction tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Streams/StreamCursorJsonConverterTests.cs | Adds StreamCursor STJ converter tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/State/StateManagerTests.cs | Adds state manager recovery/clear behavior tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/State/StateManagerExtensionsTests.cs | Adds grain extension guard + registration tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/State/StateManagerRegistrationExtensionsTests.cs | Adds DI registration tests for state manager factories |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxTests.cs | Adds functional outbox behavior tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxJsonConverterFactoryTests.cs | Adds Outbox STJ converter tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxMessageEnvelopeJsonConverterFactoryTests.cs | Adds envelope STJ converter tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxSequenceTokenJsonConverterTests.cs | Adds token STJ converter tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxReconcilerTests.cs | Adds reconciler attempt-count tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Tests/Outboxes/OutboxPostmanServiceCollectionExtensionsTests.cs | Adds DI registration tests for outbox postmen |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Streams.EventHubs.Tests/Egil.Orleans.Messaging.Streams.EventHubs.Tests.csproj | Adds Event Hubs extension test project |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Streams.EventHubs.Tests/EventHubs/EnrichedEventHubAdapterTests.cs | Adds adapter behavior tests (traceparent/enqueue time) |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Streams.EventHubs.Tests/EventHubs/EnrichedEventHubAdapterExtensionsTests.cs | Adds configurator extension tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.Streams.EventHubs.Tests/EventHubs/EnrichedEventHubSequenceTokenJsonConverterTests.cs | Adds enriched token STJ tests via registry |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.State.AzureStorage.Tests/Egil.Orleans.Messaging.State.AzureStorage.Tests.csproj | Adds Azure Storage state manager test project |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.State.AzureStorage.Tests/AzureStorage/AzureStorageStateManagerTests.cs | Adds Azure failure classification + recovery tests |
| Egil.Orleans.Messaging/test/Egil.Orleans.Messaging.State.AzureStorage.Tests/AzureStorage/AzureStorageStateManagerRegistrationExtensionsTests.cs | Adds Azure DI registration tests |
| AGENTS.md | Adds Conventional Commit scope om for new project |
| .gitignore | Fixes verify pattern + ignores local .dotnet artifacts |
| .github/workflows/egil-orleans-messaging-ci.yml | Adds CI (build/pack/test/validate/release) for project |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| /// committed-state fence that justifies the wrapper over raw | ||
| /// <see cref="IPersistentState{TState}"/>. | ||
| /// </remarks> | ||
| T State { get; } |
| private static bool IsEquivalent(T persisted, T attempted) | ||
| { | ||
| if (persisted is VersionedState persistedVersioned | ||
| && attempted is VersionedState attemptedVersioned) | ||
| { | ||
| return persistedVersioned.Version == attemptedVersioned.Version; | ||
| } | ||
|
|
||
| return persisted.Equals(attempted); | ||
| } |
| var streamNamespace = ReadRequiredStringProperty(ref reader, nameof(StreamCursor.StreamNamespace)); | ||
|
|
||
| ReadRequiredPropertyName(ref reader, nameof(StreamCursor.Token)); | ||
| if (!reader.Read()) |
| case nameof(OutboxSequenceToken.SequenceNumber): | ||
| sequenceNumber = reader.GetInt64(); | ||
| hasSequenceNumber = true; | ||
| break; | ||
| case nameof(OutboxSequenceToken.Sender): | ||
| sender = reader.TokenType is JsonTokenType.Null | ||
| ? null | ||
| : GrainIdJsonConverter.Instance.Read(ref reader, typeof(GrainId), options); | ||
| break; | ||
| case nameof(OutboxSequenceToken.Timestamp): | ||
| timestamp = reader.GetDateTimeOffset(); | ||
| hasTimestamp = true; | ||
| break; | ||
| case nameof(OutboxSequenceToken.Epoch): | ||
| epoch = reader.GetDateTimeOffset(); | ||
| hasEpoch = true; | ||
| break; |
Fall back to TimeProvider.System when Orleans rehydrates an outbox without its non-serialized clock. Document custom-provider re-registration and cover the Orleans deep-copy path so the first mutation no longer fails.
Release the drain gate between background dispatch and non-interleaving reconciliation, and let a foreground drain finish any pending batch before taking a new snapshot. This prevents non-reentrant grain calls from blocking the reconciliation turn while preserving direct PostAsync completion semantics.
No description provided.