diff --git a/CLAUDE.md b/CLAUDE.md index 2f719ba5..eec1155a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1009,7 +1009,7 @@ can pass all of it and still violate the boundary in production. function emits one `pg_notify` per distinct `(stream_context, stream_purpose)` in the transition table, carrying that stream's maximum over the total `(event_tx, event_position)` order. It was `FOR EACH ROW`, which meant a 1000-event append queued 1000 notifications and an import chunk queued 5000 — all but one - per stream discarded by `OptimizingApendListenerDecorator` after being built as JSON, written to the + per stream discarded by `OptimizingAppendListenerDecorator` after being built as JSON, written to the cluster-wide async queue, sent over the wire, parsed by Jackson and fanned out to every listener. Measured on the PG16 floor, a 100k-row insert: the notification count falls from 100.000 to exactly 1, and trigger time roughly halves — 1230ms to 460ms on one run, 808ms to 369ms on another (the absolute @@ -1050,7 +1050,9 @@ can pass all of it and still violate the boundary in production. every listener has consumed, so a stalled listener does make usage accumulate monotonically across transactions — but from a base low enough that the amplification was a throughput and latency problem, not a correctness-of-operation one. -- `stream_purpose` defaults to `'default'` in the DDL, matching `EventStreamId.DEFAULT_PURPOSE`. On a database created before this alignment (default was `''`), operators doing raw SQL inserts should run `ALTER TABLE events ALTER COLUMN stream_purpose SET DEFAULT 'default';` — no data migration is needed since all events written through the library bind the purpose explicitly +- `stream_purpose` defaults to `'default'` in the DDL, matching `EventStreamId.DEFAULT_PURPOSE` — a public + constant, so an interop layer can bind the same value the library does rather than copy the literal out + of this file. On a database created before this alignment (default was `''`), operators doing raw SQL inserts should run `ALTER TABLE events ALTER COLUMN stream_purpose SET DEFAULT 'default';` — no data migration is needed since all events written through the library bind the purpose explicitly - **Idempotency keys are scoped per event stream (context + purpose), not per storage/table.** Uniqueness is enforced by the partial unique index `idx_events_stream_idempotency` on `(stream_context, stream_purpose, idempotency_key) WHERE idempotency_key IS NOT NULL` (schema validation requires it), so the same key used on two unrelated streams does not collide and dedup behaviour does not depend on how storage instances / prefixes are wired at runtime. The `idempotency_key` is persisted and surfaced on `StoredEvent` when reading (it is not exposed on the public `Event` record). A duplicate append is still silently ignored (returns an empty result). On a database created before this change (when `idempotency_key` had a table-wide `UNIQUE`), migrate with: `ALTER TABLE events DROP CONSTRAINT events_idempotency_key_key; CREATE UNIQUE INDEX idx_events_stream_idempotency ON events (stream_context, stream_purpose, idempotency_key) WHERE idempotency_key IS NOT NULL;` — no data migration is needed - **The duplicate is recognised by the index the server names, never by the message text.** Both the append and the import path go through `isIdempotencyKeyViolation`, which pairs SQLSTATE 23505 with diff --git a/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventSource.java b/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventSource.java index da47c683..fe817c74 100644 --- a/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventSource.java +++ b/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventSource.java @@ -347,9 +347,10 @@ default Stream> query ( EventQuery query ) { * maintain only one bookmark per stream. Tags can be attached to bookmarks for * additional metadata (e.g., processing status, reader state). * - * @param reader the unique name/identifier of the reader placing the bookmark + * @param reader the unique name/identifier of the reader placing the bookmark; must not be null * @param reference the event reference to bookmark (the last processed event) * @param tags optional tags to attach to the bookmark for metadata + * @throws NullPointerException if {@code reader} is null */ void placeBookmark ( String reader, EventReference reference, Tags tags ); @@ -359,8 +360,9 @@ default Stream> query ( EventQuery query ) { * Returns the last bookmarked position for the specified reader, allowing * the reader to resume processing from where it left off. * - * @param reader the unique name/identifier of the reader + * @param reader the unique name/identifier of the reader; must not be null * @return an Optional containing the bookmarked EventReference if found, empty if no bookmark exists + * @throws NullPointerException if {@code reader} is null */ Optional getBookmark ( String reader ); @@ -394,8 +396,9 @@ default Stream> query ( EventQuery query ) { * stream.query(EventQuery.matchAll()).forEach(this::processEvent); * } * - * @param reader the unique name/identifier of the reader whose bookmark should be removed + * @param reader the unique name/identifier of the reader whose bookmark should be removed; must not be null * @return an Optional containing the previous bookmarked EventReference if one existed, empty otherwise + * @throws NullPointerException if {@code reader} is null */ Optional removeBookmark ( String reader ); diff --git a/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventStreamId.java b/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventStreamId.java index aba3190a..2d409d73 100644 --- a/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventStreamId.java +++ b/sliceworkz-eventstore-api/src/main/java/org/sliceworkz/eventstore/stream/EventStreamId.java @@ -72,7 +72,21 @@ */ public record EventStreamId ( String context, String purpose ) { - private static final String DEFAULT_PURPOSE = "default"; + /** + * The purpose given to a stream created with {@link #forContext(String)} or {@link #defaultPurpose()}: + * {@code "default"}. + *

+ * Public because it is a storage-level value, not only a Java one. It is what the library binds into + * the {@code stream_purpose} column for a context that never sets a purpose, and what the PostgreSQL + * DDL carries as that column's default — so anyone writing rows by hand, building an interop layer, + * or querying the events table directly needs the exact string this library agrees on rather than a + * literal copied from documentation. + *

+ * Note that this is a compile-time constant, so a reference to it is inlined into the calling class. + * Changing it would therefore not be a drop-in replacement — but it is stored data (see the + * {@code stream_purpose} notes in the project documentation), so it is not going to change. + */ + public static final String DEFAULT_PURPOSE = "default"; /** * Creates an EventStreamId for a specific context with the default purpose. @@ -242,6 +256,10 @@ public boolean canRead ( EventStreamId actualStreamId ) { *

* This is useful for scenarios where a general stream (e.g., "customer#anyPurpose") needs to * append events to specific instances (e.g., "customer#123"). + *

+ * A wildcard concretizes nothing: {@code forContext("customer").anyPurpose()} does not + * concretize itself, or any other wildcard-purpose stream, because it supplies no purpose to fill + * the other's wildcard in with. * * @param otherStreamId the stream ID to check if this stream concretizes it * @return true if this stream ID concretizes the other stream ID, false otherwise @@ -249,7 +267,10 @@ public boolean canRead ( EventStreamId actualStreamId ) { */ public boolean concretizes ( EventStreamId otherStreamId ) { // if the other stream is of the type "#" and this is "#" - return otherStreamId.isAnyPurpose() && !otherStreamId.isAnyContext() && otherStreamId.context().equals(this.context); + return otherStreamId.isAnyPurpose() + && !otherStreamId.isAnyContext() + && !this.isAnyPurpose() // a wildcard fills in nothing + && otherStreamId.context().equals(this.context); } /** diff --git a/sliceworkz-eventstore-api/src/test/java/org/sliceworkz/eventstore/stream/EventStreamIdTest.java b/sliceworkz-eventstore-api/src/test/java/org/sliceworkz/eventstore/stream/EventStreamIdTest.java index fefbb4ea..e320cb85 100644 --- a/sliceworkz-eventstore-api/src/test/java/org/sliceworkz/eventstore/stream/EventStreamIdTest.java +++ b/sliceworkz-eventstore-api/src/test/java/org/sliceworkz/eventstore/stream/EventStreamIdTest.java @@ -72,6 +72,29 @@ void testConcretizes ( ) { assertFalse(EventStreamId.forContext("customer").withPurpose("123").concretizes(EventStreamId.anyContext().withPurpose("123"))); } + /** + * A wildcard supplies no purpose, so it concretizes nothing — not another wildcard-purpose stream + * in the same context, and not itself. The javadoc has always said so ("This stream has a specific + * purpose (not a wildcard)"); the implementation used to check only the other three conditions. + */ + @Test + void testWildcardPurposeConcretizesNothing ( ) { + EventStreamId customerAnyPurpose = EventStreamId.forContext("customer").anyPurpose(); + EventStreamId otherCustomerAnyPurpose = EventStreamId.forContext("customer").anyPurpose(); + EventStreamId supplierAnyPurpose = EventStreamId.forContext("supplier").anyPurpose(); + + assertFalse(customerAnyPurpose.concretizes(otherCustomerAnyPurpose)); + assertFalse(customerAnyPurpose.concretizes(customerAnyPurpose)); + assertFalse(customerAnyPurpose.concretizes(supplierAnyPurpose)); + assertFalse(EventStreamId.anyContext().concretizes(customerAnyPurpose)); + + // canAppendTo is unaffected: two equal wildcard streams already matched on equals(), and the + // append path rejects a read-only target regardless -- which is why this was never reachable + // through append(). + assertTrue(customerAnyPurpose.canAppendTo(otherCustomerAnyPurpose)); + assertTrue(customerAnyPurpose.isReadOnly()); + } + @Test void testToString ( ) { EventStreamId i = EventStreamId.anyContext(); diff --git a/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/EventStoreImpl.java b/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/EventStoreImpl.java index fe5bd93f..d282f633 100644 --- a/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/EventStoreImpl.java +++ b/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/EventStoreImpl.java @@ -19,6 +19,7 @@ import java.util.Collections; import java.util.List; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -607,7 +608,7 @@ public void close ( ) { @Override public void subscribe(EventStreamEventuallyConsistentAppendListener eventuallyConsistentSubscriber) { checkStoreNotClosed(); - this.eventuallyConsistentSubscribers.add(new OptimizingApendListenerDecorator(eventuallyConsistentSubscriber)); + this.eventuallyConsistentSubscribers.add(new OptimizingAppendListenerDecorator(eventuallyConsistentSubscriber)); subscribeToStorage(); } @@ -797,11 +798,11 @@ public void notify(BookmarkPlacedNotification bookmarkPlaced) { if ( closed.get() ) { return; // see notify(AppendsToEventStoreNotification) } - LOGGER.debug("Must asynchronously notify {} eventually consistent bookmark listeners on {} of update for {} to {}", eventuallyConsistentSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark()); + LOGGER.debug("Must asynchronously notify {} eventually consistent bookmark listeners on {} of update for {} to {}", bookmarkSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark()); // schedule for execution on different thread to notify/interrupt any waiting eventual consistent processors submitOrDropIfClosed(executorServiceForBookmarkUpdates, ( ) -> { - LOGGER.debug("Notifying {} eventually consistent bookmark listeners on {} of update for {} to {}", eventuallyConsistentSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark()); + LOGGER.debug("Notifying {} eventually consistent bookmark listeners on {} of update for {} to {}", bookmarkSubscribers.size(), eventStreamId, bookmarkPlaced.reader(), bookmarkPlaced.bookmark()); bookmarkSubscribers.stream().forEach(s->s.bookmarkUpdated(bookmarkPlaced.reader(), bookmarkPlaced.bookmark())); }); } @@ -826,6 +827,7 @@ private void submitOrDropIfClosed ( ExecutorService executorService, Runnable no @Override public void placeBookmark(String reader, EventReference reference, Tags tags) { checkStoreNotClosed(); + requireReader(reader); meterBookmarkPlace.increment(); eventStorage.bookmark(reader, reference, tags); } @@ -843,8 +845,29 @@ public Optional removeBookmark(String reader) { @Override public Optional getBookmark(String reader) { checkStoreNotClosed(); + requireReader(reader); meterBookmarkGet.increment(); - return eventStorage.getBookmark(reader.toString()); + return eventStorage.getBookmark(reader); + } + + /** + * Rejects a null reader name here, rather than letting one reach a backend. + *

+ * A reader name is the whole identity of a bookmark, so a null one is a programming error in + * every case. What it is not is a defined one: the backends disagree about it. The in-memory + * stores keep bookmarks in a {@link java.util.HashMap}, which accepts a null key happily and + * stores a bookmark nobody can name; PostgreSQL has {@code reader TEXT PRIMARY KEY}, so a null + * reaches the database and comes back as a not-null violation on a place, and as a silent + * no-op on a remove and an empty {@code Optional} on a get, because {@code WHERE reader = NULL} + * matches nothing. + *

+ * Two of those paths used to be guarded accidentally, by a {@code reader.toString()} on a value + * already typed {@code String}. Removing those calls (they convert nothing) would have taken the + * guard with them, so the check is made explicit and uniform instead — one place, same answer on + * every backend, and it names the parameter. + */ + private void requireReader ( String reader ) { + Objects.requireNonNull(reader, "reader must not be null"); } @Override diff --git a/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecorator.java b/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecorator.java similarity index 92% rename from sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecorator.java rename to sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecorator.java index 7edb4331..934d0ffb 100644 --- a/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecorator.java +++ b/sliceworkz-eventstore-impl/src/main/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecorator.java @@ -43,10 +43,15 @@ * The optimization leverages the return value of {@link EventStreamEventuallyConsistentAppendListener#eventsAppended(EventReference)} * to track what the delegate listener has actually processed, allowing it to skip notifications * for event references already handled. + *

+ * Internal. Every listener passed to {@code subscribe(...)} is wrapped in one of these + * by the store itself, so there is no reason for a caller to name this class — wrapping a listener + * before subscribing it only gets it wrapped twice. It lives in the implementation package the + * ServiceLoader exists to hide, and carries no compatibility promise. * * @see EventStreamEventuallyConsistentAppendListener */ -public class OptimizingApendListenerDecorator implements EventStreamEventuallyConsistentAppendListener { +public class OptimizingAppendListenerDecorator implements EventStreamEventuallyConsistentAppendListener { private final EventStreamEventuallyConsistentAppendListener delegate; private final ReentrantLock lock; private final AtomicReference lastNotifiedReference; @@ -58,7 +63,7 @@ public class OptimizingApendListenerDecorator implements EventStreamEventuallyCo * * @param delegate the listener to decorate with optimization logic; must not be null */ - public OptimizingApendListenerDecorator(EventStreamEventuallyConsistentAppendListener delegate) { + public OptimizingAppendListenerDecorator(EventStreamEventuallyConsistentAppendListener delegate) { this.delegate = delegate; this.lock = new ReentrantLock(); this.lastNotifiedReference = new AtomicReference<>(); diff --git a/sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecoratorTest.java b/sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecoratorTest.java similarity index 94% rename from sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecoratorTest.java rename to sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecoratorTest.java index 09504876..8772aeaf 100644 --- a/sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingApendListenerDecoratorTest.java +++ b/sliceworkz-eventstore-impl/src/test/java/org/sliceworkz/eventstore/impl/OptimizingAppendListenerDecoratorTest.java @@ -32,7 +32,7 @@ import org.sliceworkz.eventstore.events.EventReference; import org.sliceworkz.eventstore.stream.EventStreamEventuallyConsistentAppendListener; -class OptimizingApendListenerDecoratorTest { +class OptimizingAppendListenerDecoratorTest { /** * Enough pairs to catch a lost notification; the window is narrow, so the count is what makes this @@ -59,7 +59,7 @@ void testLaterOfTwoSimultaneousNotificationsIsNeverLost ( ) throws InterruptedEx try ( ExecutorService notifiers = Executors.newVirtualThreadPerTaskExecutor() ) { for ( int pair = 0; pair < RACING_PAIRS; pair++ ) { AtomicReference seenByDelegate = new AtomicReference<>(); - OptimizingApendListenerDecorator decorator = new OptimizingApendListenerDecorator( + OptimizingAppendListenerDecorator decorator = new OptimizingAppendListenerDecorator( reference -> { seenByDelegate.set(reference); return reference; @@ -105,7 +105,7 @@ void testNotificationAlreadySeenByTheListenerIsSkipped ( ) { deliveries.incrementAndGet(); return reference; }; - OptimizingApendListenerDecorator decorator = new OptimizingApendListenerDecorator(counting); + OptimizingAppendListenerDecorator decorator = new OptimizingAppendListenerDecorator(counting); EventReference reference = reference(1); decorator.eventsAppended(reference); @@ -125,7 +125,7 @@ void testSlowListenerDoesNotBlockTheNotifyingThread ( ) throws Exception { CountDownLatch releaseListener = new CountDownLatch(1); AtomicReference seenByDelegate = new AtomicReference<>(); - OptimizingApendListenerDecorator decorator = new OptimizingApendListenerDecorator(reference -> { + OptimizingAppendListenerDecorator decorator = new OptimizingAppendListenerDecorator(reference -> { listenerEntered.countDown(); try { releaseListener.await(5, TimeUnit.SECONDS); diff --git a/sliceworkz-eventstore-infra-postgres/src/main/java/org/sliceworkz/eventstore/infra/postgres/PostgresEventStorageImpl.java b/sliceworkz-eventstore-infra-postgres/src/main/java/org/sliceworkz/eventstore/infra/postgres/PostgresEventStorageImpl.java index 211cdb29..5a2846ef 100644 --- a/sliceworkz-eventstore-infra-postgres/src/main/java/org/sliceworkz/eventstore/infra/postgres/PostgresEventStorageImpl.java +++ b/sliceworkz-eventstore-infra-postgres/src/main/java/org/sliceworkz/eventstore/infra/postgres/PostgresEventStorageImpl.java @@ -2215,7 +2215,7 @@ ON CONFLICT (reader) } try ( PreparedStatement stmt = writeConnection.prepareStatement(sql) ) { - stmt.setString(1, reader.toString()); + stmt.setString(1, reader); stmt.setLong(2, eventReference == null?0:eventReference.position()); stmt.setString(3, eventReference == null?"0":Long.toUnsignedString(eventReference.tx())); stmt.setString(4, eventReference==null?null:eventReference.id().value()); @@ -2258,8 +2258,8 @@ public void removeBookmark(String reader ) { writeConnection.setAutoCommit(false); try ( PreparedStatement stmt = writeConnection.prepareStatement(sql) ) { - stmt.setString(1, reader.toString()); - + stmt.setString(1, reader); + stmt.executeUpdate(); writeConnection.commit(); }