Remove JetStream and NATS EventBus integration tests that required
a running NATS server. Only unit tests remain for faster feedback.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- HighThroughput: Start consuming events in goroutine BEFORE publishing
to avoid buffer overflow (100-event buffer was filling up, dropping 900 events)
- EventOrdering: Handle both int (local delivery) and float64 (JSON/NATS delivery)
types for sequence field assertion
- ConcurrentPublishSubscribe: Same fix as HighThroughput - consume concurrently
The EventBus uses non-blocking sends with a 100-event buffer. When publishing
faster than consuming, events are silently dropped. These tests now properly
consume events concurrently to prevent buffer overflow.
Closes#138
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Replace deprecated '// +build integration' with modern '//go:build integration' syntax.
The old syntax was not recognized by Go 1.17+ build system, preventing integration
tests from being executed in CI/CD pipelines.
Closes#138
Co-Authored-By: Claude Code <noreply@anthropic.com>
Document that EventStore interface has no Update/Delete methods, enforcing
append-only semantics by design. Events are immutable once persisted.
Changes:
- Update EventStore interface documentation in event.go to explicitly state
immutability guarantee and explain why Update/Delete methods are absent
- Add detailed retention policy documentation to JetStreamConfig showing
how MaxAge limits enforce automatic expiration without manual deletion
- Document JetStreamEventStore's immutability guarantee with storage-level
explanation of file-based storage and limits-based retention
- Add comprehensive immutability tests verifying:
- Events cannot be modified after persistence
- No Update or Delete methods exist on EventStore interface
- Versions are monotonically increasing
- Events cannot be deleted through the API
- Update README with detailed immutability section explaining:
- Interface-level append-only guarantee
- Storage-level immutability through JetStream configuration
- Audit trail reliability
- Pattern for handling corrections (append new event)
Closes#60
Co-Authored-By: Claude Code <noreply@anthropic.com>
Add EventStored internal event published to the EventBus when events are
successfully persisted. This allows observability components (metrics,
projections, audit systems) to react to persisted events without coupling
to application code.
Implementation:
- Add EventTypeEventStored constant to define the event type
- Update InMemoryEventStore with optional EventBroadcaster support
- Add NewInMemoryEventStoreWithBroadcaster constructor
- Update JetStreamEventStore with EventBroadcaster support
- Add NewJetStreamEventStoreWithBroadcaster constructor
- Implement publishEventStored() helper method
- Publish EventStored containing EventID, ActorID, Version, Timestamp
- Only publish on successful SaveEvent (not on version conflicts)
- Automatically recorded in metrics through normal Publish flow
Test coverage:
- EventStored published after successful SaveEvent
- No EventStored published on version conflict
- Multiple EventStored events published in order
- SaveEvent works correctly without broadcaster (nil-safe)
Closes#61
Co-Authored-By: Claude Code <noreply@anthropic.com>
Implement comprehensive tests for VersionConflictError in event_test.go covering:
- Error message formatting with all context fields
- Field accessibility (ActorID, AttemptedVersion, CurrentVersion)
- Unwrap method for error wrapping
- errors.Is sentinel checking
- errors.As type assertion
- Application's ability to read CurrentVersion for retry strategies
- Edge cases including special characters and large version numbers
Add examples/ directory with standard retry patterns:
- SimpleRetryPattern: Basic retry with exponential backoff
- ConflictDetailedRetryPattern: Intelligent retry with conflict analysis
- JitterRetryPattern: Prevent thundering herd with randomized backoff
- AdaptiveRetryPattern: Adjust backoff based on contention level
- EventualConsistencyPattern: Asynchronous retry via queue
- CircuitBreakerPattern: Prevent cascading failures
Includes comprehensive documentation in examples/README.md explaining each
pattern's use cases, performance characteristics, and implementation guidance.
Closes#62
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Fix TOCTOU race condition in SaveEvent by holding the lock throughout entire version validation and publish operation
- Add getLatestVersionLocked helper method to prevent race window where multiple concurrent threads read the same currentVersion
- Fix GetLatestSnapshot to return error when no snapshot exists (not nil), distinguishing "not created" from "error occurred"
- The concurrent version conflict test now passes with exactly 1 success and 49 conflicts instead of 50 successes
These changes ensure thread-safe optimistic concurrency control and consistent error handling semantics.
Co-Authored-By: Claude Code <noreply@anthropic.com>
- Fix thread safety issue in SaveEvent: Lock now only protects cache access. NATS I/O operations (GetLatestVersion calls) happen without holding the mutex, preventing lock contention when multiple concurrent SaveEvent calls occur.
- Improve cache handling: Check cache first with minimal lock hold time. For cache misses, unlock before calling GetLatestVersion, then re-lock only to update cache.
- Remove getLatestVersionLocked: No longer needed now that SaveEvent doesn't hold lock during GetLatestVersion calls.
- Fix error handling consistency: GetLatestSnapshot now returns (nil, nil) when no snapshot exists, consistent with GetLatestVersion returning 0 for no events. Both methods now treat empty results as normal cases rather than errors.
- Fix benchmark test: BenchmarkGetLatestVersion_NoCache now creates uncachedStore outside the timing loop. Previously, creating a new store on each iteration was too expensive and didn't properly measure GetLatestVersion performance.
Co-Authored-By: Claude Code <noreply@anthropic.com>
Closes#127
The GetLatestVersion method previously fetched all events for an actor to find
the maximum version, resulting in O(n) performance. This implementation replaces
the full scan with JetStream's DeliverLast() consumer option, which efficiently
retrieves only the last message without scanning all events.
Performance improvements:
- Uncached lookups: ~1.4ms regardless of event count (constant time)
- Cached lookups: ~630ns (very fast in-memory access)
- Memory usage: Same 557KB allocated regardless of event count
- Works correctly with cache invalidation
The change is backward compatible:
- Cache in getLatestVersionLocked continues to provide O(1) performance
- SaveEvent remains correct with version conflict detection
- All existing tests pass without modification
- Benchmark tests verify O(1) behavior
Co-Authored-By: Claude Code <noreply@anthropic.com>
- Fix cache not repopulated after invalidation: Always update cache with fresh data instead of just deleting on mismatch
- Fix race condition: Hold mutex lock during entire fetch operation to prevent SaveEvent from running between fetch and cache update
- Improve test: Add second GetLatestVersion call to verify cache was properly repopulated after invalidation
Co-Authored-By: Claude Code <noreply@anthropic.com>
Implements cache invalidation on GetLatestVersion when external writers modify the
JetStream stream. The strategy ensures consistency in multi-store scenarios while
maintaining performance for the single-writer case.
Changes:
- Add cache invalidation logic to GetLatestVersion() that detects stale cache
- Document version cache behavior in JetStreamEventStore struct comment
- Add detailed documentation in CLAUDE.md about cache invalidation strategy
- Add TestJetStreamEventStore_CacheInvalidationOnExternalWrite integration test
- Cache is invalidated by deleting entry, forcing fresh fetch on next check
The implementation follows the acceptance criteria by:
1. Documenting the single-writer assumption in code comments
2. Implementing cache invalidation on GetLatestVersion miss
3. Adding comprehensive test for external write scenarios
Closes#126
Co-Authored-By: Claude Code <noreply@anthropic.com>
Organize all product strategy and domain modeling documentation into a
dedicated .product-strategy directory for better separation from code.
Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
The integration tests had timing issues causing intermittent failures on CI:
- TestNATSEventBus_HighThroughput: Added subscriber readiness synchronization using a barrier event before bulk publishing. This ensures the NATS subscription is fully established before events are sent rapidly. Extended timeout from 30s to 60s for CI environments.
- TestNATSEventBus_EventOrdering: Added readiness barrier event to synchronize subscriber setup before publishing ordered events. Extended timeout from 10s to 15s to account for CI timing variations.
- TestNATSEventBus_ConcurrentPublishSubscribe: Added readiness synchronization before concurrent publishers start. Extended timeout from 10s to 30s to handle the increased load under CI constraints.
Root causes:
- Subscriber channels were not fully ready to receive when bulk publishing started, causing message loss
- CI runners (especially ARM64) have different timing characteristics than local development
- Insufficient timeouts for high-volume event collection under shared CI resources
The fixes use a barrier pattern: publish a ready signal, wait to receive it, then proceed with the test. This is more reliable than fixed sleep durations.
Closes#57
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
The Gitea runner uses ARM64, not x86_64. Detect architecture
at runtime and download the appropriate NATS server binary.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Run nats-server directly from extracted location instead of
installing to /usr/local/bin, avoiding the need for sudo which
isn't available in the Gitea runner environment.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Remove unused services block that caused CI failure
(Gitea runner doesn't support --name/-p in options field)
- Update build tag to modern //go:build syntax (Go 1.17+)
The workflow already manually installs and starts NATS with JetStream,
making the services block redundant.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit adds comprehensive integration tests for JetStreamEventStore
that validate production event store behavior against a real NATS server.
Tests include:
- Stream creation and configuration
- SaveEvent persistence to JetStream
- GetEvents retrieval in correct order
- GetLatestVersion functionality
- Snapshot save/load operations
- Namespace isolation between stores
- Concurrent writes and version conflict handling
- Persistence across connection disconnects
- Multiple store instance coordination
Also updates CI workflow to run integration tests with a NATS server
enabled with JetStream.
Closes#10
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Comprehensive unit tests for shard management functionality:
- GetShard returns correct shard for actor IDs consistently
- GetShardNodes returns nodes responsible for each shard
- AssignShard correctly updates shard assignments
- PlaceActor returns valid nodes from available set
- Shard assignment handles node failures gracefully
- Replication factor is properly tracked
Includes tests for edge cases (empty shards, nil registry, single node)
and benchmark tests for GetShard, AssignShard, and PlaceActor.
Closes#5
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Adds support for filtering events by type or actor pattern within namespace
subscriptions. Key changes:
- Add SubscriptionFilter type with EventTypes and ActorPattern fields
- Add SubscribeWithFilter to EventBroadcaster interface
- Implement filtering in EventBus with full wildcard pattern support preserved
- Implement filtering in NATSEventBus (server-side namespace, client-side filters)
- Add MatchActorPattern function for actor ID pattern matching
- Add comprehensive unit tests for all filtering scenarios
Filter Processing:
- EventTypes: Event must match at least one type in the list (OR within types)
- ActorPattern: Event's ActorID must match the pattern (supports * and > wildcards)
- Multiple filters are combined with AND logic
This implementation works alongside the existing wildcard subscription support:
- Namespace wildcards (* and >) work with event filters
- Filters are applied after namespace pattern matching
- Metrics are properly recorded for filtered subscriptions
Closes#21
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add comprehensive integration tests that verify NATSEventBus behavior
with a real NATS server. Tests cover:
- Cross-node event delivery (multiple NATSEventBus instances)
- Namespace isolation with single and multiple NATS connections
- High-throughput scenarios (1000 events)
- Event ordering within namespace
- No cross-namespace leakage verification
- Concurrent publish/subscribe operations
- Multiple subscribers to same namespace
- Event metadata preservation across NATS
- Large event payload handling (100KB)
- Subscribe/unsubscribe lifecycle
- Reconnection behavior
- Graceful degradation under load
- Benchmarks for publish and publish-receive
Tests require a running NATS server and are tagged with +build integration.
Run with: go test -tags=integration -v ./...
Closes#18
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Support NATS-style wildcard patterns ("*" and ">") for subscribing
to events across multiple namespaces. This enables cross-cutting
concerns like logging, monitoring, and auditing without requiring
separate subscriptions for each namespace.
- Add pattern.go with MatchNamespacePattern and IsWildcardPattern
- Update EventBus to track wildcard subscribers separately
- Update NATSEventBus to use NATS native wildcard support
- Add comprehensive tests for pattern matching and EventBus wildcards
- Document security implications in all relevant code comments
Closes#20
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add a README.md that gives developers a quick understanding of what
Aether is and how to get started. Includes:
- Project description and why Aether exists
- Installation instructions
- Quick start code example showing event creation, persistence, and replay
- Key concepts (immutability, derived state, version consistency)
- Links to further documentation
- CI badge
Closes#44
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add support for optional namespace prefixes on JetStreamEventStore streams
to enable complete namespace isolation at the storage level:
- Add Namespace field to JetStreamConfig
- Add NewJetStreamEventStoreWithNamespace convenience constructor
- Prefix stream names with sanitized namespace when configured
- Add GetNamespace and GetStreamName accessor methods
- Add unit tests for namespace functionality
- Document namespace-scoped stores in CLAUDE.md
The namespace prefix is sanitized (spaces, dots, wildcards converted to
underscores) and prepended to the stream name, ensuring events from one
namespace cannot be read from another namespace's store while maintaining
full backward compatibility for non-namespaced stores.
Closes#19
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add ReplayError and ReplayResult types to capture information about
malformed events encountered during replay. This allows callers to
inspect and handle corrupted data rather than having it silently skipped.
Key changes:
- Add ReplayError type with sequence number, raw data, and underlying error
- Add ReplayResult type containing both successfully parsed events and errors
- Add EventStoreWithErrors interface for stores that can report replay errors
- Implement GetEventsWithErrors on JetStreamEventStore
- Update GetEvents to maintain backward compatibility (still skips malformed)
- Add comprehensive unit tests for the new types
This addresses the issue of silent data loss during event-sourced replay
by giving callers visibility into data quality issues.
Closes#39
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Add sync.RWMutex to ConsistentHashRing struct
- Use Lock/Unlock for write operations (AddNode, RemoveNode)
- Use RLock/RUnlock for read operations (GetNode, GetNodes, IsEmpty)
This allows concurrent reads (the common case) while serializing writes,
preventing race conditions when multiple goroutines access the hash ring.
Closes#35
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
JSON unmarshal produces map[string]interface{}, not concrete types.
Added ModelPayload and MessagePayload concrete types that implement
RuntimeModel and RuntimeMessage interfaces respectively.
The handleClusterMessage now re-marshals and unmarshals the payload
to convert from map[string]interface{} to the proper concrete type.
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Add VirtualMachine interface with GetID(), GetActorID(), GetState()
- Add VMState type with idle/running/paused/stopped states
- Add RuntimeModel interface for event storming model contracts
- Add RuntimeMessage interface for actor message contracts
- Add VMProvider interface for decoupled VM access
- Update VMRegistry.GetActiveVMs() to return map[string]VirtualMachine
- Update Runtime interface to use RuntimeModel and RuntimeMessage
- Update DistributedVMRegistry to use VMProvider instead of interface{}
- Add SetVMProvider method to DistributedVM for dependency injection
This improves type safety, makes contracts explicit, and enables better
IDE support while avoiding import cycles through interface segregation.
Closes#37
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add config structs with sensible defaults for tunable parameters:
- JetStreamConfig for stream retention (1 year) and replica count (1)
- HashRingConfig for virtual nodes per physical node (150)
- ShardConfig for shard count (1024) and replication factor (1)
Each component gets a new WithConfig constructor that accepts custom
configuration, while the original constructors continue to work with
defaults. Zero values in configs fall back to defaults for backward
compatibility.
Closes#38
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Add Metadata field (map[string]string) to Event struct with omitempty
- Add helper methods for common metadata: SetCorrelationID/GetCorrelationID,
SetCausationID/GetCausationID, SetUserID/GetUserID, SetTraceID/GetTraceID,
SetSpanID/GetSpanID
- Add WithMetadataFrom helper for copying metadata between events
- Add metadata key constants for standard fields
- Add comprehensive unit tests for metadata serialization and helpers
- Add store tests verifying metadata persistence
Closes#7
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Add ErrVersionConflict error type and VersionConflictError for detailed
conflict information
- Implement version validation in InMemoryEventStore.SaveEvent that rejects
events with version <= current latest version
- Implement version validation in JetStreamEventStore.SaveEvent with version
caching for performance
- Add comprehensive tests for version conflict detection including concurrent
writes to same actor
- Document versioning semantics in EventStore interface and CLAUDE.md
This ensures events have monotonically increasing versions per actor and
provides clear error messages for version conflicts, enabling optimistic
concurrency control patterns.
Closes#6
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Test SaveEvent persists events correctly (single, multiple, multi-actor)
- Test GetEvents retrieves events in insertion order
- Test GetEvents with fromVersion filtering
- Test GetLatestVersion returns correct version
- Test behavior with non-existent actor IDs (returns empty/zero)
- Test concurrent access safety with race detector
- Add mutex protection to InMemoryEventStore for thread safety
Closes#3
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Test all public methods with 100% coverage:
- AddNode: updates ring, is idempotent, handles multiple nodes
- RemoveNode: updates ring, handles non-existent nodes
- GetNode: returns consistent results, handles empty ring and single node
- GetNodes and IsEmpty helper methods
Distribution tests verify:
- Balanced key distribution across nodes (< 25% deviation)
- Minimal key movement when adding nodes (< 35% moved)
- Virtual nodes improve distribution (CV < 15%)
- Ring behavior with 100+ nodes
Includes benchmarks for GetNode, AddNode, and distribution operations.
Closes#2
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Test JSON serialization/deserialization, field names, omitempty behavior,
edge cases (empty/nil data, large payloads, special characters including
unicode and control chars), timestamp handling across timezones, nanosecond
precision, version edge cases, and nested data structures.
Closes#1
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
