Skip to content

CLAUDE.md

Latest commit

 

History

History
168 lines (120 loc) · 53.3 KB

CLAUDE.md

File metadata and controls

168 lines (120 loc) · 53.3 KB
Raw

CLAUDE.md

This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.

Build & Test Commands

# Build (Debug — no doc requirements)
dotnet build RayTree.slnx

# Build (Release — used by CI)
dotnet build RayTree.slnx -c Release

# Run all unit tests (no Docker required)
dotnet test tests/RayTree.Core.Tests
dotnet test tests/RayTree.Plugins.InMemory.Tests
dotnet test tests/RayTree.EntityFrameworkCore.Tests
dotnet test tests/RayTree.OpenTelemetry.Tests
dotnet test tests/RayTree.Plugins.Compressors.{Brotli,Gzip,Lz4}.Tests
dotnet test tests/RayTree.Plugins.Serializers.{Json,MessagePack,Protobuf}.Tests

# Run a single test by name
dotnet test tests/RayTree.Core.Tests --filter "FullyQualifiedName~NoSerializer"

# Run integration tests (requires Docker — spins up containers via Testcontainers)
dotnet test tests/RayTree.Plugins.PostgreSQL.Tests
dotnet test tests/RayTree.Plugins.RabbitMQ.Tests
dotnet test tests/RayTree.Plugins.Kafka.Tests
dotnet test tests/RayTree.Plugins.Deduplication.Redis.Tests

TreatWarningsAsErrors=true is global. Nullable warnings are always errors. All new public code must satisfy these constraints.

Centralized package versions live in Directory.Packages.props. Add new packages there; reference them in .csproj without a version attribute.

Architecture Overview

RayTree is a modular .NET 10 entity change-tracking library built on the outbox pattern. All change tracking flows through a single EntityChangeTracker that acts as the unified host for both the publisher and subscriber pipelines:

EntityChangeTracker
  → IOutbox (persist change + entity state)
  → OutboxPublisherService (polls outbox, serializes, publishes)
  → IQueuePublisher (broker-specific)
      ↓ MessageEnvelope (meta headers + byte[] Payload)
  → IQueueConsumer
  → ChangeSubscriber (internal: dedup, decompress, deserialize, dispatch)
  → ChangeHandlerAsync<TEntity>(EntityChange<TEntity>, CancellationToken)

Core (src/RayTree.Core)

  • EntityChangeTracker — the single runtime host. Constructor-injects a ChangePublisher (required) and a ChangeSubscriber? (optional). internal InitializeAsync() starts publisher loops (via ChangePublisher.InitializeAsync()) and initializes all consumer connections (ChangeSubscriber.InitializeAsync() parallelises consumer init via Task.WhenAll so a single slow consumer — e.g. Kafka WaitForTopic against a missing topic — does not block the others) — called automatically by Build()/BuildAsync(), never by callers directly. Public lifecycle surface: StartAsync(CancellationToken) starts all shared and isolated consumer loops (stores tasks internally); StopAsync() awaits those tasks (swallows OperationCanceledException); RunCleanupAsync(TimeSpan retentionPeriod, CancellationToken) iterates all registered outboxes and calls CleanupPublishedAsync, returning total rows deleted. TrackXxxAsync writes to the internal outbox via GetOutbox(entityType). Publisher and Subscriber are internal — plugin assemblies access them via InternalsVisibleTo; callers use the public API instead.

  • ChangeTrackingBuilder / IChangeTrackingBuilder — unified fluent builder for both sides. Accepts an optional ILoggerFactory? constructor parameter; null normalizes to NullLoggerFactory.Instance, so existing call-sites that omit it continue to work. Global factories (UseOutbox<T>, UseSerializer<T>, etc.) apply to all entity types. UseSerializer/UseCompressor at the global level forward to both the publisher factory and the subscriber's global instance. UseSubscriberOptions and UseDeduplicationStore configure the subscriber globally. Per-entity overrides live inside .ForEntity<TEntity>(Action<IEntityBuilder<TEntity>>) which exposes both publisher methods (UseOutbox, UsePublisher, UseSerializer, UseCompressor, UseRepository) and subscriber methods (UseConsumer, OnInsert, OnUpdate, OnDelete, OnChange, UseSubscriberOptions). Build() / BuildAsync() produce a fully initialized EntityChangeTracker with the subscriber already attached.

  • IEntityBuilder<TEntity> — generic per-entity configuration interface. Publisher side: UseOutbox, UsePublisher(IQueuePublisher), UseSerializer, UseCompressor, UseRepository. Subscriber side: UseConsumer(IQueueConsumer), UseSubscriberOptions, OnInsert, OnUpdate, OnDelete, OnChange. where TEntity : class is required because subscriber handler registration is typed.

  • ChangePublisher — owns all publisher-side plugin registrations (IOutbox, IQueuePublisher, IChangeSerializer, IChangeCompressor, IRepository per entity type) in ConcurrentDictionary<Type, …>, and manages the OutboxPublisherService instances. Constructor signature: (ILoggerFactory loggerFactory, RayTreeMeter meter) — both are required non-null parameters. Exposes Meter as internal for EntityChangeTracker and tests (visible to RayTree.Core.Tests via InternalsVisibleTo). InitializeAsync() initializes repositories, outboxes, publishers, then starts one OutboxPublisherService per registered entity type, passing the meter to each. Parallel to ChangeSubscriber on the subscriber side.

  • OutboxPublisherService — background polling loop per entity type: reads unpublished changes → serialize → compress → wrap in MessageEnvelope → publish → mark published → rotate outbox (see below). Constructor signature: (ChangePublisher, Type entityType, OutboxPublisherOptions, ILoggerFactory, RayTreeMeter meter) — the meter is a required non-null parameter. Caches _entityTag = RayTreeMeter.EntityTag(entityType) in the constructor for hot-path reuse. Emits raytree.outbox.batch.size per batch; per attempt records raytree.outbox.publish.duration; on success records raytree.outbox.messages.published + raytree.outbox.publish.attempts + raytree.outbox.lag.duration; on exhaustion records raytree.outbox.messages.failed + raytree.outbox.publish.attempts (so retry-shape histograms reflect worst cases too). PublishChangeAsync records raytree.outbox.payload.size (bytes) with the envelope's payload length. MaybeRunCleanupAsync records raytree.outbox.records.cleaned and raytree.outbox.stale_unpublished.removed. When OutboxPublisherOptions.UseNotificationChannel = true, uses FallbackPollingInterval instead of PollingInterval as the inter-batch sleep, demoting the service to a safety-net role (catching anything the NOTIFY fast-path misses) while NotificationBasedPublisher handles normal delivery. Each batch is published in parallel via Parallel.ForEachAsync bounded by MaxPublishConcurrency. When a batch is full (changes.Count == BatchSize), the inter-batch sleep is skipped entirely to drain the backlog immediately. Logs start/stop at Information; batch errors at Error; per-retry warnings at Warning; exhausted retries at Error. After each batch it calls MaybeRunCleanupAsync, which fires once on the first tick and then respects OutboxPublisherOptions.CleanupInterval (default 1 h). Rotation logs: start at Debug; records deleted at Information; nothing to delete at Debug; stale unpublished records found at Warning; errors at Error (isolated — a cleanup failure does not abort the publish loop).

  • OutboxPublisherOptions — PollingInterval (default 5 s), BatchSize (default 100), MaxPublishConcurrency (default 1 — sequential; increase for throughput when message ordering within a partition is not required), MaxRetryCount (default 3), RetryDelay (default 1 s), UseNotificationChannel / NotificationChannel / FallbackPollingInterval for PostgreSQL NOTIFY/LISTEN. Rotation options: CleanupRetentionPeriod (default 7 days — how old a published record must be before deletion), CleanupInterval (default 1 h — how often rotation runs), StaleUnpublishedThreshold (default null — opt-in; when set, unpublished records older than this value are also removed with a Warning log so operators notice stuck queues).

  • MessageEnvelope — the only thing that crosses the queue boundary. Contains metadata fields plus byte[] Payload (already serialized+compressed entity state). Serialization happens on the publisher side; deserialization on the subscriber side.

  • IChangeSerializer / IChangeCompressor — stream-based interfaces. Serializers handle EntityChange<TEntity> with typed State. NoOpCompressorPlugin (built-in) is a passthrough for testing.

  • IOutbox — ten methods: InitializeAsync, WriteAsync<TEntity>, two GetUnpublishedAsync<TEntity> overloads (by batch size; or by ChangeType?, DateTime? since, batch size), MarkPublishedAsync, TryClaimForPublishingAsync(id) (atomically sets published = TRUE WHERE published = FALSE, returns true if this caller claimed it), RevertClaimAsync(id) (sets published = FALSE — used to undo a claim when publish fails so the fallback loop can retry), GetByIdAsync<TEntity>, GetPendingCountAsync(Type entityType) (returns long — count of unpublished records for the given entity type; used by the raytree.outbox.pending observable gauge; throws InvalidOperationException if called with the wrong entity type for a typed implementation), CleanupPublishedAsync(retentionPeriod), CleanupStaleUnpublishedAsync(staleThreshold). Both cleanup methods return the count of rows deleted.

  • RayTreeMeter (src/RayTree.Core/Telemetry) — owns a single System.Diagnostics.Metrics.Meter("RayTree", <assembly-version>) and all 14 outbox/subscriber instruments (counters, histograms, and the raytree.outbox.pending observable gauge). The four raytree.connection.* instruments were removed — connection recovery is now observed via logs only. Created by ChangeTrackingBuilder when UseMeter is not called; EntityChangeTracker disposes it only when it created it (the ownsMeter flag tracks this). When the caller supplies their own meter via UseMeter(RayTreeMeter), the caller owns disposal. RegisterPendingGauge(Func<IEnumerable<(Type, IOutbox)>>) (now internal) registers the raytree.outbox.pending observable gauge; the callback samples IOutbox.GetPendingCountAsync per registered entity type and caches results for DefaultPendingCacheTtl = 10s to bound DB round-trips. All instruments are tagged with entity_type; change-specific instruments add change_type; skipped-message counters add reason. The class exposes internal instrument properties and internal emit methods (RecordPublishSuccess, RecordPublishFailure, RecordPayloadSize, RecordBatchSize, RegisterPendingGauge) consumed by EntityChangeTracker, OutboxPublisherService, ChangeSubscriber, and the IVT-privileged NotificationBasedPublisher. The public surface is now construct-and-observe only: MeterName, the constructors, DefaultPendingCacheTtl, and Dispose() — all metric emission is internal. RayTreeMeter.MeterName is a public constant equal to "RayTree". See docs/opentelemetry-metrics.md for the full instrument inventory and unit conventions (s for durations, By for bytes).

  • ChangeSubscriber — internal implementation detail owned by EntityChangeTracker. Receives MessageEnvelope from an IQueueConsumer, deduplicates on CorrelationId via IDeduplicationStore, decompresses and deserializes the payload using reflection (DeserializeCoreAsync<TEntity> via MethodInfo.MakeGenericMethod), then dispatches to registered ChangeHandlerAsync<TEntity> handlers with retry. ConsumeFromConsumerAsync and ConsumeFromQueueAsync use Parallel.ForEachAsync bounded by SubscriberOptions.MaxDegreeOfParallelism (default 1 — sequential, preserves per-partition ordering). When all handlers succeed, calls MaybeDedupCleanupAsync to periodically evict old dedup entries (gated by SubscriberOptions.DeduplicationCleanupInterval); a SemaphoreSlim gate ensures only one concurrent invocation runs cleanup at a time. When a handler exhausts retries and throws (SkipOnFailure = false), reverts the dedup mark via RevertProcessedAsync before rethrowing — this ensures the redelivered message can be retried rather than silently dropped by a persistent dedup store. Still public for direct low-level usage (e.g., standalone subscriber tests), but not the primary API. Constructor signature: (ILogger<ChangeSubscriber> logger, RayTreeMeter meter, IDeduplicationStore? dedupStore = null, SubscriberOptions? options = null) — logger and meter are required non-null parameters. Emits raytree.subscriber.messages.skipped (with reason="unknown_type" or reason="no_handler"), raytree.subscriber.messages.deduplicated, raytree.subscriber.messages.processed, and raytree.subscriber.lag.duration from ProcessMessageAsync. InvokeWithRetryAsync records raytree.subscriber.processing.duration per attempt (success or failure), raytree.subscriber.handler.attempts on every completed dispatch (success or failure), and raytree.subscriber.handler.failures on exhaustion. Logs unknown entity types at Warning; dedup hits at Debug; no-handler-match at Debug; successful message dispatch at Debug; dedup revert on handler failure at Warning; retry attempts at Warning; SkipOnFailure drops at Error; dedup cleanup errors at Error.

  • ChangeSubscriberBuilder / IChangeSubscriberBuilder — standalone subscriber-only builder. Produces a ChangeSubscriber via Build(). ChangeTrackingBuilder composes one internally; BuildInternal() calls _subscriberBuilder.Build() and passes the result to the EntityChangeTracker constructor.

Publisher-side plugins

Package What it provides
RayTree.Plugins.PostgreSQL PostgreSqlOutbox<TEntity> — stores changes as flat columns (one column per entity property via EntityColumnMapper). Constructor: PostgreSqlOutbox<TEntity>(PostgreSqlOutboxOptions, ILoggerFactory) — both params required. PostgreSqlRepository<TEntity> constructor: PostgreSqlRepository<TEntity>(PostgreSqlRepositoryOptions, ILoggerFactory) — both params required. Builder extension methods accept ILoggerFactory? loggerFactory = null and default to NullLoggerFactory.Instance. EntityColumnMapper honours System.ComponentModel.DataAnnotations / Schema attributes: [NotMapped] excludes a property; [Column("name")] overrides the column name suffix (the state_ prefix is always kept to avoid collisions with outbox metadata columns); [Column(TypeName = "JSONB")] sets the PostgreSQL type verbatim; [Required] forces NOT NULL on reference types; [MaxLength(n)]/[StringLength(n)] emits VARCHAR(n) instead of TEXT; [Table("name")] on the entity class is used as the base name when deriving default outbox/source table names; [Key] (one or more properties) identifies the business primary key — PostgreSqlRepository uses these for INSERT/UPDATE/DELETE/SELECT and adds a UNIQUE index on the corresponding state_* columns in the source table; for composite keys pair [Key] with [Column(Order = n)] to control column order. 1D arrays of primitive types are automatically mapped to the corresponding PostgreSQL array column type: int[] → INTEGER[], long[] → BIGINT[], bool[] → BOOLEAN[], string[] → TEXT[], Guid[] → UUID[], float[] → REAL[], double[] → DOUBLE PRECISION[], decimal[] → NUMERIC[], DateTime[]/DateTimeOffset[] → TIMESTAMPTZ[], short[]/byte[]/sbyte[] → SMALLINT[]; nullable-element arrays (e.g. int?[]) strip the nullable wrapper before mapping the element type. Multi-dimensional arrays are not supported — declare the column type explicitly via [Column(TypeName = "...")] if needed. When reading values back, EntityColumnMapper.ConvertFromDb first attempts a direct CLR assignability check (Npgsql returns the correct array type natively) and falls back to Convert.ChangeType for scalar numeric coercions. Both CleanupPublishedAsync and CleanupStaleUnpublishedAsync delete in batches (PostgreSqlOutboxOptions.CleanupBatchSize, default 1000) using a DELETE … WHERE id IN (SELECT id … LIMIT @BatchSize) loop to avoid large single-statement locks and WAL spikes. InitializeAsync manages schema automatically — no flag required, always active. Fresh table path: single CREATE TABLE IF NOT EXISTS (columns + indexes). Existing table path: column diff via SchemaMigrator (adds missing columns with ALTER TABLE … ADD COLUMN IF NOT EXISTS; guards NOT NULL without default on non-empty tables by throwing InvalidOperationException; logs Warning for orphan columns and type mismatches) + index diff via IndexMigrator (creates missing indexes; drops and recreates indexes whose definition changed — uniqueness, column order, or WHERE clause; logs Warning for orphan indexes). Internal infrastructure: SchemaInspector (static — TableExistsAsync, GetColumnsAsync via information_schema.columns, GetIndexesAsync via pg_index catalog using unnest(indkey::smallint[]) WITH ORDINALITY for ordered columns and pg_get_expr for WHERE, ExecuteDdlAsync, TableHasRowsAsync); SchemaMigrator (column diff, parameterised delegate for DDL generation and orphan filter); IndexMigrator (index diff with schema-qualified DROP INDEX IF EXISTS public.{name}; WHERE clause comparison is case-insensitive and trimmed); PostgreSqlTypeNormalizer (maps information_schema type fields to canonical DDL strings). NotificationBasedPublisher — NOTIFY/LISTEN fast-path with polling fallback; bounded by NotificationBasedPublisherOptions.MaxConcurrentNotifications (default 16) via a SemaphoreSlim in OnNotification; fallback polling uses Parallel.ForEachAsync with MaxPublishConcurrency (default 1 — sequential). Logs LISTEN connection loss at Warning (once, on the first unhealthy tick), recovery at Information, and claim contention (record already taken by another publisher) at Debug.
RayTree.Plugins.InMemory InMemoryQueue implements both IQueuePublisher and IQueueConsumer via Channel<MessageEnvelope>. Use for tests and local dev.
RayTree.Plugins.Kafka KafkaPublisher + KafkaConsumer. Publisher key: KafkaPublisherOptions.KeySelector (Func<MessageEnvelope, string>) selects the Kafka partition key for each message. Default: envelope => $"{EntityType}:{EntityId}" — all changes for the same entity land on the same partition, preserving per-entity ordering. Override to shard by any envelope field (e.g. tenant, aggregate root). Consumer uses a dedicated background thread (channel-based) because Confluent.Kafka requires all Consume/Commit/Seek calls on one thread. KafkaConsumer(KafkaConsumerOptions, ILoggerFactory) — both params required. KafkaConsumerOptions.AckAfterHandler (default false) defers the offset commit; subscriber posts the ConsumeResult plus a Commit/SeekBack action through an internal post-handler channel that the poll thread drains at the top of each iteration (when items are queued, the next Consume() uses TimeSpan.Zero so commits don't wait a full poll cycle). AcknowledgeAsync → Commit; NegativeAcknowledgeAsync → Seek(TopicPartitionOffset) so the failed message is redelivered in the same consumer's lifetime, not just on restart. Parse-failure path always commits immediately to avoid poison-pilling the partition. Requires SubscriberOptions.MaxDegreeOfParallelism = 1 per partition when AckAfterHandler = true. Topic wait (opt-in, both options classes): WaitForTopic (bool, default false), TopicWaitInterval (TimeSpan, default 5 s), TopicWaitTimeout (TimeSpan?, default null — unlimited). When WaitForTopic = true, InitializeAsync probes the configured Topic via IAdminClient.GetMetadata and retries while the response indicates the topic is not yet available — defined as: empty Topics collection, per-topic ErrorCode.UnknownTopicOrPart, per-topic ErrorCode.LeaderNotAvailable (transient during cluster bootstrap / partition-leader election), OR a transient transport-level KafkaException with Error.IsFatal == false and Error.Code in {Local_Transport, Local_AllBrokersDown, Local_Resolve, Local_TimedOut} — the dominant startup-ordering case where the broker pod has not yet finished starting. All other broker errors and fatal KafkaExceptions propagate immediately (authorization failures, unrecoverable client state). Each GetMetadata call is bounded by a fixed KafkaTopicProbe.MetadataCallTimeout (1 s) decoupled from TopicWaitInterval so cancellation latency and shutdown thread-pool occupancy are bounded regardless of the interval. The publisher routes the probe through the lazy GetProducerAsync path used by both InitializeAsync and PublishAsync, so callers that publish without explicit init still benefit; the consumer probes before allocating the native IConsumer handle AND re-checks cancellationToken.ThrowIfCancellationRequested() immediately after the probe to prevent handle leaks when cancellation arrives during a slow probe. The publisher uses two SemaphoreSlim instances: a one-shot _probeSemaphore gated by volatile bool _probeCompleted (so steady-state callers skip it entirely once the probe has run) and a separate microsecond-long _buildSemaphore for the producer build — splitting the two prevents concurrent PublishAsync callers from serializing behind a multi-second probe. KafkaPublisher.Dispose is idempotent (volatile bool _disposed guard) and uses a SafeRelease helper that swallows ObjectDisposedException from in-flight Release() calls during a Dispose-during-init race. KafkaPublisher(KafkaPublisherOptions, ILoggerFactory? = null) accepts an optional logger factory (null → NullLoggerFactory.Instance) so the probe can log progress; both builder extensions — KafkaBuilderExtensions.UseKafka(configure, loggerFactory) and KafkaSubscriberExtensions.UseKafka<TEntity>(configure, loggerFactory) — expose an optional ILoggerFactory? parameter so the documented fluent API can forward host logging (without this on the subscriber side the probe would silently drop all logs). Probe logging cadence matches TopologyProbe: first miss Information, subsequent misses Debug, recovery Information, timeout exhaustion Error. Use this in microservice deployments where the topic owner pod comes up after the consumer/publisher. Auto-create caveat: brokers with auto.create.topics.enable=true (the default on many distributions) create the topic in response to the probe itself, masking real misconfiguration — set the broker option to false in deployments that rely on this feature; the integration test container uses WithEnvironment("KAFKA_AUTO_CREATE_TOPICS_ENABLE", "false") for the same reason.
RayTree.Plugins.RabbitMQ RabbitMqPublisher + RabbitMqConsumer. Routing key: RabbitMqPublisherOptions.RoutingKeySelector (Func<MessageEnvelope, string>) selects the AMQP routing key for each message. Default: "{RoutingKey}.{EntityType}.{changeType}" (e.g. change.Order.insert) — consumers bind queues with wildcard patterns such as change.Order.* or change.*.insert. The default delegate reads RoutingKey at call time so changing that property after construction is always reflected; set a custom delegate to route by tenant, aggregate root, or any envelope field. RabbitMqPublisher(RabbitMqPublisherOptions, ILoggerFactory?) — options required, logger factory optional (null → NullLoggerFactory.Instance); UseRabbitMq(configure, loggerFactory) mirrors the same shape. Consumer uses AsyncEventingBasicConsumer buffered via Channel<MessageEnvelope>. RabbitMqConsumer(RabbitMqConsumerOptions) — options only; no logger. Message-receive errors silently NACK and requeue without logging (acknowledged exception to the logging placement rule — NACK/requeue is the correct recovery action and no context is available at that point). RabbitMqConsumerOptions.AckAfterHandler (default false) defers the broker ACK until after ChangeSubscriber confirms handler success — delivery tag is stashed in MessageEnvelope.Metadata via the internal RabbitMqEnvelopeMetadata accessor; AcknowledgeAsync issues BasicAckAsync; NegativeAcknowledgeAsync issues BasicNackAsync(requeue: true). Topology wait (opt-in, both options classes): WaitForTopology (bool, default false), TopologyWaitInterval (TimeSpan, default 5 s), TopologyWaitTimeout (TimeSpan?, default null — unlimited). When WaitForTopology = true, InitializeAsync probes externally-owned topology via AMQP passive declares (ExchangeDeclarePassiveAsync / QueueDeclarePassiveAsync) and retries only on NOT_FOUND (404) until the topology appears, the cancellation token is cancelled, or TopologyWaitTimeout elapses (rethrowing the last NOT_FOUND). Other channel- and connection-level errors (PRECONDITION_FAILED, ACCESS_REFUSED, etc.) propagate immediately. Each probe attempt uses a fresh channel from the existing connection because RabbitMQ closes the channel on any channel-level exception. The publisher probes when DeclareExchange = false; the consumer probes the queue when DeclareQueue = false and the binding-target exchange when ExchangeName is non-empty. Probe progress is logged by the publisher (first miss Information, subsequent misses Debug, recovery Information, timeout Error via TopologyProbe); the consumer's no-logger exception still holds, so consumer-side probes log nothing. Use this in microservice deployments where one service owns the topology and others connect later without strict startup ordering.
RayTree.Plugins.Serializers.* JSON, MessagePack, Protobuf — each in its own package.
RayTree.Plugins.Compressors.* Gzip, Brotli, LZ4 — each in its own package.

Handler dispatch modes

IEntityBuilder<TEntity> exposes two consumer-binding methods that select the dispatch mode and fork the fluent chain into a mode-specific builder:

  • UseConsumer(IQueueConsumer) → ISharedHandlerBuilder<TEntity> — Shared mode. All handlers registered on the returned builder share a single broker delivery. They run sequentially in registration order. Dedup key: correlationId. If any handler exhausts retries with SkipOnFailure = false, the message-level dedup mark is reverted so redelivery will re-run every handler. Callers must chain UseSerializer/UseCompressor on IEntityBuilder before calling UseConsumer (those methods are not on the post-fork builder).
  • UseConsumerFactory(Func<string, IQueueConsumer>) → IIsolatedHandlerBuilder<TEntity> — Isolated mode. Each named handler (OnInsert("name", handler)) gets its own broker subscription via the factory. Dedup key: $"{correlationId}:{handlerName}" — per-handler isolation. The framework starts one consume loop per (entity type, handler name) pair inside tracker.StartAsync. Handler names are stable deployment identifiers: renaming one is equivalent to creating a new subscription.

Handler registration methods (OnInsert, OnUpdate, OnDelete, OnChange) exist only on the post-fork builders, not on IEntityBuilder<TEntity> — the compiler prevents registering handlers before binding a consumer, and prevents mixing anonymous and named overloads.

ChangeType is required on every handler registration. OnChange(changeType, handler) takes a non-nullable ChangeType; the previous wildcard null form (which fired for every change type) was removed. Each handler binds to exactly one of Insert / Update / Delete. To react to multiple change types with the same logic, register the same delegate once per type — the dispatch matcher is now a strict equality check (h.ChangeType == envelope.ChangeType), and there is no implicit fall-through. This applies uniformly to IEntitySubscriberBuilder<TEntity>, ISharedHandlerBuilder<TEntity>, IIsolatedHandlerBuilder<TEntity>, and ChangeSubscriber.OnChange<TEntity> / RegisterIsolatedHandler<TEntity>. The HandlerRegistration.ChangeType field is correspondingly non-nullable.

Builder implementation classes (src/RayTree.Core/Handling): SharedHandlerBuilder<TEntity> delegates to EntitySubscriberBuilder<TEntity>; IsolatedHandlerBuilder<TEntity> accumulates (handlerName, changeType, handler) tuples and validates them (unique (action, name) pairs; factory returns distinct non-null instances) at Build() time.

Delivery-guarantee mode (Shared and Isolated): by default, RabbitMqConsumer and KafkaConsumer ACK/commit the broker delivery before handing the message to the subscriber (at-most-once — lowest latency, no broker-driven redelivery on crash). Opt in to at-least-once per consumer by setting RabbitMqConsumerOptions.AckAfterHandler = true or KafkaConsumerOptions.AckAfterHandler = true; the ACK is then deferred until ChangeSubscriber confirms all handlers completed successfully. On handler retry-exhaustion with SkipOnFailure = false, the consumer's NegativeAcknowledgeAsync requeues (RabbitMQ BasicNack(requeue: true)) or seeks back (Kafka Seek on the poll thread — redelivery happens in the same consumer process, no restart needed). Implementation: IQueueConsumer exposes default-no-op AcknowledgeAsync / NegativeAcknowledgeAsync methods; existing implementations inherit the no-ops and behave as at-most-once. Broker-private correlation state (delivery tag for RMQ, ConsumeResult for Kafka) travels with the envelope via MessageEnvelope.Metadata (lazy-allocated dict; consumed via take-on-read accessors so a double-ack is a silent no-op rather than a broker error). Kafka caveat: set MaxDegreeOfParallelism = 1 per partition when AckAfterHandler = true — out-of-order commits could advance the offset past in-flight messages. The ChangeSubscriber.ConsumeFromConsumerAsync(IQueueConsumer) overload is the one that participates in the Ack lifecycle; the custom-reader overload ConsumeFromQueueAsync<TQueue> stays at-most-once by design (no IQueueConsumer to dispatch against).

Subscriber-side (src/RayTree.Core/Handling)

  • ChangeSubscriber — see Core section above.
  • Handler signature: ChangeHandlerAsync<TEntity>(EntityChange<TEntity> change, CancellationToken ct). change.State is the fully-typed entity; it is null when no serializer is registered for that entity type.
  • IDeduplicationStore — three methods: TryMarkProcessedAsync(correlationId) (atomic add, returns false if already present — the primary dedup gate), RevertProcessedAsync(correlationId) (removes the entry — called on handler failure so re-delivered messages can be retried), CleanupAsync(retentionPeriod) (evicts entries older than the retention window). InMemoryDeduplicationStore is the default (process-local, cleared on restart). Use RedisDeduplicationStore for distributed deployments where dedup state must survive restarts.
  • SubscriberOptions — MaxDegreeOfParallelism (default 1 — sequential, preserves per-partition ordering; increase for throughput when handlers are order-independent), MaxRetries (retry attempts after the first call), RetryDelay, SkipOnFailure, DeduplicationRetention (default 24 h — how old a processed CorrelationId must be before cleanup eligibility), DeduplicationCleanupInterval (default 1 h — how often ChangeSubscriber calls IDeduplicationStore.CleanupAsync after a successful message).

.NET Generic Host integration (src/RayTree.Hosting)

  • AddChangeTracking(services, configuration, configure) — the primary registration path. Registers EntityChangeTracker as a singleton (publisher + subscriber configured together), RayTreeMeter as a singleton, and ChangeTrackingHostedService as a hosted service. Resolves ILoggerFactory from the DI container and passes it to new ChangeTrackingBuilder(loggerFactory) — no explicit logging setup is required from the caller. The DI-registered RayTreeMeter is fed back into the builder via UseMeter(sp.GetRequiredService<RayTreeMeter>()) so the same instance is shared between the tracker and any consumer that injects RayTreeMeter directly (e.g., custom instrumentation). Publisher loops start during Build() (via internal InitializeAsync). The hosted service calls tracker.StartAsync(ct) on app startup to start consumer loops. Publisher options are bound from ChangeTracking:Publisher; subscriber options from ChangeTracking:Subscriber.
  • ChangeTrackingHostedService — thin IHostedService adapter. StartAsync delegates to tracker.StartAsync(_cts.Token); StopAsync cancels the token then awaits tracker.StopAsync(), logging graceful shutdown at Information. No loop management logic lives here — the tracker owns that entirely. Constructor requires (EntityChangeTracker, ILogger<ChangeTrackingHostedService>) — both are auto-wired by DI.

OpenTelemetry integration (src/RayTree.OpenTelemetry)

Peer assembly — referenced only by applications that want OTel SDK wiring. RayTree.Core and RayTree.Hosting have zero OpenTelemetry.* dependencies, so apps that ignore this package receive no transitive OTel closure.

  • RayTreeInstrumentation — public static class with public const string MeterName = "RayTree";. Use this constant in custom OTel views/filters instead of hard-coding the literal.
  • MeterProviderBuilderExtensions.AddRayTreeMetrics(this MeterProviderBuilder builder) — thin pass-through that calls builder.AddMeter(RayTreeInstrumentation.MeterName). Does not configure exporters, views, or histogram bucket boundaries — callers retain full control.

All durations are emitted in seconds (s) per OTel semantic conventions; bytes use By. Suggested histogram bucket boundaries for each instrument are documented in docs/opentelemetry-metrics.md. Tests use an inline CapturingExporter : BaseExporter<Metric> + BaseExportingMetricReader instead of pulling in OpenTelemetry.Exporter.InMemory.

EF Core integration (src/RayTree.EntityFrameworkCore)

EntityChangeInterceptor hooks into SaveChangesAsync to automatically call TrackInsertAsync/TrackUpdateAsync/TrackDeleteAsync based on EF change tracker state.

Connection recovery (per-plugin)

Connection recovery is behavior + logs only — there are no connection metrics. The recovery-owning plugins each define their own options type carrying the exponential-backoff shape: PostgresConnectionRecoveryOptions (src/RayTree.Plugins.PostgreSQL/Resilience) and KafkaConnectionRecoveryOptions (src/RayTree.Plugins.Kafka). Both are near-verbatim copies of the same six members (Enabled, InitialDelay, MaxDelay, Factor, JitterFraction, MaxAttempts) with per-field init guards and a Validate() cross-field check. There is no shared Core options type (the former RayTree.Core.Resilience.ConnectionRecoveryOptions was removed) and no Hosting binding (the former ChangeTrackingRecoveryKeys + services.Configure<ConnectionRecoveryOptions>(...) calls were removed); callers configure recovery per plugin in the UseKafka / UsePostgreSqlOutbox configure lambda.

Recovery shape per plugin (logs only — disconnect/recovery are no longer counted):

  • postgres.notification (NotificationBasedPublisher.ListenLoopAsync) — single long-lived NpgsqlConnection as a local await using var (no class field). On a fault classified by the shared PostgresFault.IsConnectionFault static helper (uses Npgsql.PostgresErrorCodes constants — admin/crash shutdown, 08xxx connection_exception family, transient NpgsqlException, SocketException/IOException inner, ObjectDisposedException), logs a Warning once per fault cycle, then runs an inline exponential-backoff loop bounded by NotificationBasedPublisherOptions.ConnectionRecovery (a PostgresConnectionRecoveryOptions) that disposes the connection, opens a fresh one, reissues LISTEN. On reconnect logs Information; on MaxAttempts exhaustion logs Error and exits the loop. Fallback polling continues throughout.

  • postgres.outbox — observation only, no rebuild. Two seams classify faults:

    • OutboxPublisherService.ProcessBatchAsync: HandleBatchError consults IOutbox.IsConnectionFault + ConnectionComponent in the existing batch-error catch. On a classified fault: demote per-batch Error log to Warning, set _unhealthySince. On the first subsequent successful batch: log an Information "outbox connection recovered" entry via EmitOutboxRecovered.
    • NotificationBasedPublisher.FallbackPollingLoopAsync: ProcessSingleOutboxAsync does the same classification per outbox, keyed on entity type via a ConcurrentDictionary<Type, FallbackOutboxState> so each outbox tracks its own transitions independently.

    IOutbox exposes three default-implemented members (IsConnectionFault, ConnectionComponent, ConnectionEndpoint) that PostgreSqlOutbox<TEntity> overrides via PostgresFault. The polling cadence is the natural retry — no rebuild loop. Non-fault exceptions preserve the original Error path. InMemoryOutbox and third-party IOutbox implementations inherit no-op defaults — no breaking change.

  • kafka.publisher — SetErrorHandler callback flips an atomic Interlocked.CompareExchange flag (_faultTicks UTC-ticks) and logs a Warning. The callback does no locking and no disposal (avoids the documented Confluent.Kafka footgun of disposing inside the native callback). The next PublishAsync enters GetProducerAsync under a single _buildLock, disposes the dead producer on a normal call thread, re-runs the WaitForTopic probe (when enabled), builds a fresh producer, and logs Information. No inner backoff loop — the outbox-publisher retry loop is the outer cadence. State: 2 fields (_producer, _faultTicks) + 1 lock.

  • kafka.consumer — fatal KafkaException on the dedicated poll thread → drops pending deferred-ack actions (broker redelivers via at-least-once contract on the new consumer's join) → disposes the dying consumer → runs an inline exponential-backoff RebuildConsumer on the same poll thread (re-runs WaitForTopic probe + Subscribe), bounded by KafkaConsumerOptions.ConnectionRecovery (a KafkaConnectionRecoveryOptions). Backoff math is a small private static helper (Task.Delay(...).GetAwaiter().GetResult() is safe here — the poll thread is dedicated with no SyncContext). On exhaustion completes the buffer channel with the fatal exception.

  • rabbitmq.publisher / rabbitmq.consumer — observe, don't own. RabbitMQ.Client.AutomaticRecoveryEnabled (library default — RayTree does not disable it) performs the actual rebuild. Publisher subscribes to ConnectionShutdownAsync (Warning when Initiator != Application), RecoverySucceededAsync (Information + duration), ConnectionRecoveryErrorAsync (Information only — per-internal-attempt). The publisher constructor no longer takes a RayTreeMeter. Consumer no longer subscribes to recovery events at all (its subscriptions existed solely to feed the removed metrics) and its constructor no longer takes a RayTreeMeter — it is silent for recovery, consistent with its no-logger posture. RabbitMqPublisherOptions / RabbitMqConsumerOptions do not expose ConnectionRecovery — the library's NetworkRecoveryInterval controls timing, exposed via ConnectionFactory not through RayTree.

The retry math (~20 LoC: exponential backoff with ±JitterFraction jitter, capped at MaxDelay, optionally bounded by MaxAttempts) is duplicated across the two plugins that need it (Postgres LISTEN, Kafka consumer), as are the Validate()d options records. This is intentional — extracting a shared helper/type would require either InternalsVisibleTo exposure to plugins (architectural smell — plugins must consume Core via public API only) or a public-API commitment to a specific retry shape. Two short copies is cheaper than either of those.

Key Design Decisions

  • Unified builder: IChangeTrackingBuilder.ForEntity<TEntity>() takes Action<IEntityBuilder<TEntity>> where IEntityBuilder<TEntity> covers both publisher and subscriber configuration. The where TEntity : class constraint is required by the subscriber handler registration. Value types cannot be entity types. UseSerializer/UseCompressor at the global level forward the factory's output to the subscriber's global instance by calling factory(typeof(object)) — this works correctly when the factory ignores the type parameter (the common case).
  • Tracker as thin coordinator: EntityChangeTracker composes a ChangePublisher (required) and a ChangeSubscriber? (optional) via constructor injection. ChangeTrackingBuilder.BuildInternal() creates the ChangePublisher, registers all plugins on it, then builds the subscriber via _subscriberBuilder.Build() and passes both to the EntityChangeTracker constructor. Publisher and Subscriber are internal; plugin assemblies (RayTree.EntityFrameworkCore, RayTree.Plugins.PostgreSQL) and test projects access them via InternalsVisibleTo entries in RayTree.Core.csproj. External callers use only the tracker's public surface (TrackXxxAsync, StartAsync, StopAsync, RunCleanupAsync, Meter).
  • Publisher loop lifetime: OutboxPublisherService instances are created and started inside the tracker's internal InitializeAsync(), called synchronously by Build() (or asynchronously by BuildAsync()). Consumer loop lifetime is separate: tracker.StartAsync(CancellationToken) starts all shared and isolated loops; tracker.StopAsync() awaits them. In production ChangeTrackingHostedService calls these; in tests or standalone apps call them directly. ChangeTrackingHostedService is a thin adapter — no loop management logic of its own.
  • Reflection for generic dispatch: OutboxPublisherService, NotificationBasedPublisher, and ChangeSubscriber all use MethodInfo.MakeGenericMethod to invoke serializer/deserializer methods with the runtime entity type. Return types are declared as the non-generic base (Task<EntityChange>) so the async upcast works cleanly.
  • PostgreSQL outbox schema: Each entity type gets its own outbox table. Entity properties are stored as flat columns (not JSON), derived via EntityColumnMapper.GetColumns(typeof(TEntity)). By default column names are state_<snake_case> and the table name is <snake_case>_outbox. Both are customisable via System.ComponentModel.DataAnnotations / Schema attributes on the entity class — see the RayTree.Plugins.PostgreSQL plugin row above for the full attribute reference. EntityColumnMapper.GetTableName(Type) encapsulates the [Table]-aware table-name logic and is the single place both PostgreSqlOutbox and PostgreSqlRepository use for their defaults. EntityColumnMapper.ToPostgresType(Type) maps CLR types to PostgreSQL column types; 1D array properties (e.g. int[]) are emitted as the corresponding PostgreSQL array type (e.g. INTEGER[]) — see the plugin row above for the full mapping table. EntityColumnMapper.ConvertFromDb(object value, Type targetType) is the shared read-path helper used by both PostgreSqlOutbox.ReadEntityChange and PostgreSqlRepository.MapEntity; it short-circuits via IsAssignableFrom (covering arrays and exact-type matches) before falling back to Convert.ChangeType for scalar numeric coercions. InitializeAsync branches on table existence: if the table does not exist, one CREATE TABLE IF NOT EXISTS creates all columns and indexes; if it exists, column diff + index diff are applied (see SchemaMigrator and IndexMigrator in the plugin row above). Three outbox indexes are declared in the schema: idx_*_outbox_unpublished — partial on (published, timestamp) WHERE published = FALSE, used by GetUnpublishedAsync; idx_*_outbox_cleanup — partial on (timestamp) WHERE published = TRUE, used by CleanupPublishedAsync; idx_*_outbox_entity — on (entity_type, published, timestamp), used by the filtered GetUnpublishedAsync overload. IndexMigrator.ApplyDiffAsync keeps these in sync with the live database on every startup by comparing against pg_index catalog data and applying DROP + CREATE for changed definitions.
  • PostgreSQL primary key resolution: EntityColumnMapper.GetKeyProperties(Type) returns the ordered list of key properties for an entity. It first looks for properties annotated with [Key]; multiple [Key] properties form a composite key ordered by [Column(Order)] then by declaration order. If no [Key] is found it falls back to the Id convention property. If neither exists it throws InvalidOperationException at construction time (fail-fast). PostgreSqlRepository uses key properties to build INSERT, WHERE (UPDATE/DELETE/SELECT), and the source-table UNIQUE index. IRepository<TEntity>.GetByIdAsync takes object[] keyValues — one value per key property in the same order.
  • Outbox rotation is part of the publisher loop, not a separate service: OutboxPublisherService.MaybeRunCleanupAsync runs inline after each batch in the same polling goroutine. It fires eagerly on the first tick (cleans up stale data from before startup), then gates subsequent runs on CleanupInterval. _lastCleanup is only advanced when both cleanup operations succeed; a failure leaves _lastCleanup unchanged so the next tick retries immediately, giving operators fast feedback via repeated error logs rather than silently waiting a full interval. This keeps rotation within the tracker's lifecycle — no extra hosted service, no external scheduler. Cleanup errors are isolated with their own try/catch so a transient DB failure does not abort the publish loop. Rotation runs sequentially (not in parallel with publishing) because the cleanup DELETE and the unpublished SELECT target disjoint row sets (published = TRUE vs published = FALSE) — no concurrency is gained, but isolation makes error handling straightforward. For ad-hoc manual rotation, call tracker.RunCleanupAsync(retentionPeriod, ct) directly.
  • NotificationBasedPublisher as a fast-path overlay: when UseNotificationChannel = true in PostgreSqlOutboxOptions, the DB trigger fires a pg_notify on every INSERT into the outbox table. NotificationBasedPublisher receives that notification, atomically claims the record via IOutbox.TryClaimForPublishingAsync (prevents races with other publishers), and publishes immediately. The fallback polling loop inside NotificationBasedPublisher runs only on the first tick (to drain records written before the listener was established) and when _listenerHealthy = false (LISTEN connection broke). OutboxPublisherService continues running but at FallbackPollingInterval cadence — it acts as a safety net for anything the notification path misses, not an equal peer. Both paths use TryClaimForPublishingAsync/RevertClaimAsync to prevent duplicate publishing without relying solely on subscriber-side deduplication.
  • Dedup mark-before-process with revert-on-failure: ChangeSubscriber calls TryMarkProcessedAsync before invoking handlers (single round-trip, prevents concurrent duplicate processing). If a handler exhausts its retries and throws, RevertProcessedAsync removes the entry before the exception propagates, so the redelivered message is accepted and retried. When SkipOnFailure = true, no revert is issued — the intentional skip is permanent. This ensures at-least-once semantics are preserved even with persistent dedup stores (e.g., Redis) that survive process restarts.
  • Kafka thread safety: KafkaConsumer keeps a single background Task.Run thread that owns all IConsumer<K,V> operations. Dispose() cancels via _disposeCts, waits up to 2×PollTimeoutMs + 200 ms for the poll task to exit, then frees the native handle.
  • Integration tests use Testcontainers: PostgreSQL, Kafka, and RabbitMQ tests require Docker. Mark test classes [NonParallelizable] when sharing a container. Use unique topic/queue names per test to avoid cross-test contamination.
  • Metrics placement rule — required meter, no null fallback: RayTreeMeter is a required non-null constructor parameter on every runtime service that emits metrics (ChangePublisher, OutboxPublisherService, ChangeSubscriber). There is no internal new RayTreeMeter() fallback in those classes — the builder layer (ChangeTrackingBuilder.BuildInternal) constructs a default meter when the caller didn't supply one via UseMeter, then injects it everywhere. This matches the logging rule: callers make a conscious choice at builder/DI level, runtime services have no hidden defaults. EntityChangeTracker tracks ownership via an ownsMeter flag and disposes the meter only when it created it; caller-supplied meters are left alone. Instrument calls are silent no-ops when no listener is attached, so opting out costs nothing at runtime.
  • OTel SDK isolation via peer assembly: RayTree.Core and RayTree.Hosting use only System.Diagnostics.Metrics (BCL) — no OpenTelemetry.* package references. RayTree.OpenTelemetry is a separate assembly with two members (RayTreeInstrumentation.MeterName + AddRayTreeMetrics) that an application opts into. Applications that don't need OTel receive zero transitive OTel dependencies; applications that do reference exactly one well-versioned dependency. This mirrors the RayTree.Hosting / RayTree.EntityFrameworkCore split.
  • Logging placement rule: NullLoggerFactory.Instance / NullLogger<T>.Instance defaults belong only in builders and builder-context extension methods (ChangeTrackingBuilder, ChangePublisherBuilder, ChangeSubscriberBuilder, KafkaSubscriberExtensions.UseKafka, RabbitMqBuilderExtensions.UseRabbitMq, RabbitMqSubscriberExtensions.UseRabbitMq, BuilderExtensions.UsePostgreSqlOutbox, RepositoryExtensions.UsePostgreSqlRepository). All runtime service classes (ChangePublisher, OutboxPublisherService, ChangeSubscriber, ChangeTrackingHostedService, KafkaConsumer, NotificationBasedPublisher, PostgreSqlOutbox<TEntity>, PostgreSqlRepository<TEntity>) require a non-nullable logger — no internal fallback. RabbitMqPublisher and KafkaPublisher both accept an optional ILoggerFactory? (null → NullLoggerFactory.Instance) to support topology-wait / topic-wait logging without forcing every caller through DI; both fluent builder extensions (KafkaBuilderExtensions.UseKafka, KafkaSubscriberExtensions.UseKafka<TEntity>) accept an optional ILoggerFactory? parameter and forward it so the consumer-side probe — whose underlying KafkaConsumer constructor already requires a non-null logger factory — is actually reachable with a real factory from the documented fluent API (previously this extension hardcoded NullLoggerFactory.Instance, silencing every probe log). Exception: RabbitMqConsumer intentionally has no logger — message-receive errors silently NACK and requeue, which is the correct broker-level recovery; no useful context is available inside the RabbitMQ delivery callback to produce a meaningful log entry. Consumer-side topology-wait probes therefore pass logger: null to TopologyProbe, preserving the exception. This ensures that callers always make a conscious choice about whether to produce log output.
  • Configuration & lifecycle logs: ChangeTrackingBuilder emits Information for each Use* registration (with {Plugin} = type name), an Information "ChangeTracker built" summary at BuildInternal time (with {EntityTypes}, {Plugins}, {HasCustomMeter}, {HasCustomDeduplicationStore}, {HasCustomLoggerFactory}), and a Debug "no meter supplied" note when it creates the default RayTreeMeter. ForEntity<TEntity> logs Information for the entity type; per-entity overrides (UseOutbox, UsePublisher, UseSerializer, UseCompressor, UseRepository, UseSubscriberOptions, UseConsumer, UseConsumerFactory, OnInsert/OnUpdate/OnDelete/OnChange) log at Debug with {EntityType}, {Override} (slot name like "Outbox" or "OnInsert:handlerName"), and {Plugin} (concrete type name). EntityChangeTracker.InitializeAsync logs Information "tracker initialization started" / "completed" bracketing two Debug sub-step entries: "publisher initialized" with {EntityTypeCount} and "consumers initialized" with {ConsumerCount}; on failure it emits one Warning "tracker initialization aborted" (no exception payload — the inner service's own Error carries the cause) before rethrowing. ChangeTrackingHostedService.StartAsync emits one Information "ChangeTracking starting" with {ConfigurationBound} (sourced from the ChangeTrackingDiContext singleton registered by AddChangeTracking). Every call is guarded by IsEnabled(...) so NullLoggerFactory produces zero allocations.

Code Style & Conventions

See AGENTS.md for all coding rules, design principles, testing conventions, nullability discipline, and AI agent workflow guidelines. Key .editorconfig conventions (naming, braces, using order) are summarized below; AGENTS.md is the canonical source.

  • Private/internal fields: _camelCase; static: PascalCase; constants: PascalCase
  • Expression-bodied members for single-expression members
  • using outside namespace; System namespaces first
  • Braces on a new line
  • Use named params, especially with multiple args of the same type

.editorconfig wins over any conflicting suggestion.

CI

.github/workflows/ci.yml has three job groups: build (compile gate, uploads compiled output as an artifact with 1-day retention), unit-tests (9-way parallel matrix, no Docker, downloads build artifact — no rebuild), integration-tests (3-way matrix: PostgreSQL / RabbitMQ / Kafka, also downloads build artifact). No job rebuilds the solution; all test jobs depend on the shared artifact from build.