Skip to content

CLAUDE.md

Latest commit

 

History

History
125 lines (87 loc) · 41.2 KB

CLAUDE.md

File metadata and controls

125 lines (87 loc) · 41.2 KB
Raw

CLAUDE.md

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

Build & Test Commands

See AGENTS.md — canonical source for build/test commands, hard constraints (TreatWarningsAsErrors, nullable, package versioning), and every coding convention. Not duplicated here.

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>(PostgreSqlOutboxOptions, ILoggerFactory) — stores changes as flat columns (one per entity property via EntityColumnMapper). PostgreSqlRepository<TEntity>(PostgreSqlRepositoryOptions, ILoggerFactory) — source/key table, same attribute handling. Both params required on both constructors; builder extensions default ILoggerFactory? to NullLoggerFactory.Instance. EntityColumnMapper honours System.ComponentModel.DataAnnotations / Schema attributes: [NotMapped] excludes a property; [Column("name")] overrides the column suffix (state_ prefix always kept); [Column(TypeName = "JSONB")] sets the PostgreSQL type verbatim; [Required] forces NOT NULL; [MaxLength(n)]/[StringLength(n)] emits VARCHAR(n); [Table("name")] sets the base table name; [Key] (composite via [Column(Order = n)]) identifies the business primary key — used for INSERT/WHERE and the source-table UNIQUE index. 1D primitive arrays map to the matching PostgreSQL array type (int[] → INTEGER[], bool[] → BOOLEAN[], Guid[] → UUID[], DateTime[]/DateTimeOffset[] → TIMESTAMPTZ[], etc. — see EntityColumnMapper.ToPostgresType for the full mapping; nullable-element arrays strip the wrapper first); multi-dimensional arrays aren't supported. Cleanup deletes in batches (CleanupBatchSize, default 1000) to avoid large single-statement locks. InitializeAsync auto-migrates schema always (fresh CREATE TABLE, or column/index diff on an existing table — guards NOT NULL-without-default on non-empty tables by throwing; logs Warning for orphan columns/indexes/type mismatches). Three outbox indexes are kept in sync automatically: idx_*_outbox_unpublished (partial, published = FALSE), idx_*_outbox_cleanup (partial, published = TRUE), idx_*_outbox_entity (entity_type, published, timestamp). NotificationBasedPublisher — optional NOTIFY/LISTEN fast path (MaxConcurrentNotifications default 16) with polling fallback. Full schema-migration internals and connection-recovery detail: src/RayTree.Plugins.PostgreSQL/README.md.
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. Default: envelope => $"{EntityType}:{EntityId}" — all changes for one entity land on the same partition, preserving per-entity ordering. Override to shard by any envelope field. Consumer uses a dedicated background thread 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 until handler success (Commit/Seek(TopicPartitionOffset) via an internal post-handler channel drained on the poll thread); requires SubscriberOptions.MaxDegreeOfParallelism = 1 per partition when enabled — commits are monotonic per partition, so concurrent workers could advance past an in-flight message. Parse failures always commit immediately to avoid poison-pilling the partition. Full delivery-guarantee mechanics: src/RayTree.Plugins.Kafka/README.md. Topic wait (opt-in, both options classes): WaitForTopic (default false), TopicWaitInterval (default 5 s), TopicWaitTimeout (default null — unlimited). When WaitForTopic = true, InitializeAsync probes the configured Topic via IAdminClient.GetMetadata and retries on transient not-yet-available conditions (empty Topics, UnknownTopicOrPart, LeaderNotAvailable, or a transient transport-level KafkaException) — the dominant startup-ordering case where the broker pod hasn't finished starting. Other errors propagate immediately. Each probe is bounded by KafkaTopicProbe.MetadataCallTimeout (1 s), decoupled from TopicWaitInterval. The publisher routes the probe through the lazy GetProducerAsync path (two separate semaphores so steady-state PublishAsync calls skip the probe once it's run); the consumer probes before allocating its native handle. Use this in microservice deployments where the topic owner pod starts after the consumer/publisher. Auto-create caveat: brokers with auto.create.topics.enable=true (common default) create the topic in response to the probe itself, masking real misconfiguration — set it false in deployments relying on this feature.
RayTree.Plugins.RabbitMQ RabbitMqPublisher + RabbitMqConsumer. Routing key: RabbitMqPublisherOptions.RoutingKeySelector (Func<MessageEnvelope, string>) selects the AMQP routing key. Default: "{RoutingKey}.{EntityType}.{changeType}" (e.g. change.Order.insert) — consumers bind with wildcard patterns like change.Order.*. Set a custom delegate to route by tenant, aggregate root, or any envelope field. RabbitMqPublisher(RabbitMqPublisherOptions, ILoggerFactory?) — options required, logger factory optional. RabbitMqConsumer(RabbitMqConsumerOptions) — options only, no logger (message-receive errors silently NACK + requeue — no useful context is available at that point, an acknowledged exception to the logging-placement rule). RabbitMqConsumerOptions.AckAfterHandler (default false) defers the broker ACK until ChangeSubscriber confirms handler success (BasicAckAsync/BasicNackAsync(requeue: true), delivery tag carried in MessageEnvelope.Metadata). Topology wait (opt-in, both options classes): WaitForTopology (default false), TopologyWaitInterval (default 5 s), TopologyWaitTimeout (default null). When true, InitializeAsync probes externally-owned topology via AMQP passive declares and retries only on NOT_FOUND (404); other errors propagate immediately. Publisher probes when DeclareExchange = false; consumer probes when DeclareQueue = false (and the binding exchange when ExchangeName is set). Full mechanics, probe diagram, and logging table: src/RayTree.Plugins.RabbitMQ/README.md#topology-wait-microservice-deployments.
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. Each recovery-owning plugin defines its own options type with the same six-field exponential-backoff shape (Enabled, InitialDelay, MaxDelay, Factor, JitterFraction, MaxAttempts): PostgresConnectionRecoveryOptions and KafkaConnectionRecoveryOptions. There is no shared Core options type and no Hosting binding — configure recovery per plugin in the UseKafka / UsePostgreSqlOutbox configure lambda.

  • PostgreSQL — NotificationBasedPublisher.ListenLoopAsync rebuilds the LISTEN connection with backoff on a classified fault (PostgresFault.IsConnectionFault); OutboxPublisherService / fallback polling only observe outbox faults (demote log level, no rebuild — pooled connections + polling cadence is the retry). Full detail: src/RayTree.Plugins.PostgreSQL/README.md#connection-recovery.
  • Kafka — publisher rebuilds its producer lazily on the next PublishAsync (GetProducerAsync under _buildLock); consumer rebuilds inline on its dedicated poll thread (RebuildConsumer), bounded by KafkaConnectionRecoveryOptions. Full detail: src/RayTree.Plugins.Kafka/README.md (Error handling section).
  • RabbitMQ — observes only; RabbitMQ.Client.AutomaticRecoveryEnabled (library default, not disabled) does the actual rebuild. Publisher logs shutdown/recovery events; consumer has no logger (design exception, see below) and is silent for recovery. RabbitMqPublisherOptions/RabbitMqConsumerOptions don't expose ConnectionRecovery — the library's NetworkRecoveryInterval controls timing.

The retry math (~20 LoC exponential backoff with jitter) is intentionally duplicated between Postgres and Kafka rather than extracted — a shared helper would need either InternalsVisibleTo exposure to plugin assemblies (architectural smell) or a public-API commitment to a specific retry shape. Two short copies is cheaper than either.

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, columns named state_<snake_case> by default (see the plugin row above for the full attribute reference, table-naming, and array-type mapping). 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.
  • 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, naming/.editorconfig conventions, and nullability discipline — canonical source, not duplicated here.

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.