From f5ee8f8c5d465989f0aeac86e077fa297902bf60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matej=20Ba=C4=8Do?= Date: Wed, 12 Aug 2026 13:06:46 +0200 Subject: [PATCH 1/5] Add handler outcomes and chunk consumption MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A handler now answers with an Outcome: Continue (processed, advance), Skip (cannot be processed and will not be retried — advance anyway), or Retry (stop here, same position, so the next run re-delivers — throwing with the exception left out). Returning anything else, or nothing, is Continue, so every existing handler keeps its meaning. consumeChunk() hands the whole poll — up to the batch setting — to the handler as one list, for work that is cheaper in bulk. One outcome answers for the chunk: the position moves past all of it or none of it. Partial progress is a seek() before returning Retry, which a run never saves over. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 17 +++ README.md | 72 ++++++++++- src/Feed/Consumer.php | 102 +++++++++++---- src/Feed/Outcome.php | 34 +++++ tests/Feed/Consumer/Base.php | 214 +++++++++++++++++++++++++++++++ tests/Feed/Unit/ConsumerTest.php | 52 ++++++++ 6 files changed, 465 insertions(+), 26 deletions(-) create mode 100644 src/Feed/Outcome.php diff --git a/CHANGELOG.md b/CHANGELOG.md index e6d7031..fcbd2eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,22 @@ # Changelog +## Unreleased + +### Consuming + +- `Outcome` — the words a handler answers with: `Continue` (processed, advance), + `Skip` (cannot be processed and will not be retried — advance anyway, said + explicitly or not at all), and `Retry` (stop here, same position, so the next + run re-delivers — throwing with the exception left out). A handler that + returns anything else, or nothing, has continued, so every existing handler + keeps its meaning. +- `Consumer::consumeChunk(callable)` — the whole poll (up to `batch` events) as + one `list` per call, for handlers whose work is cheaper in bulk. + One outcome answers for the chunk: the position moves past all of it or none + of it. A handler that made partial progress can `seek()` to the last event it + completed before returning `Retry` — a move made mid-run is never saved over. + The handler is not called for an empty poll. + ## 0.1.0 — Initial release Pull-based HTTP event feeds ([http-feeds.org](https://www.http-feeds.org/)) for diff --git a/README.md b/README.md index 4230d14..5f811eb 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,72 @@ names its feed: $consumer = new Consumer($store, new Cursor\Redis($redis), name: 'audit-log'); ``` +### Answering with an outcome + +A handler answers with an `Outcome`; anything else it returns — including +nothing — reads as `Outcome::Continue`, so a handler that simply returns is +saying "processed, move on": + +| The handler | The position | The run | +| --- | --- | --- | +| returns `Outcome::Continue` (or anything else) | advances past the event | keeps going | +| returns `Outcome::Skip` | advances past the event | keeps going | +| returns `Outcome::Retry` | stays before the event | ends, returns the count so far | +| throws | stays before the event | ends, the error is re-raised | + +`Retry` and throwing are the same decision about the feed — this event is not +handled, re-deliver it — made in different moods: throwing is for accidents +and surfaces the error, `Retry` is for failures the handler expected (a +dependency it already knows is down) and returns calmly. `Skip` is the +deliberate loss of an event, decided in code: + +```php +use Utopia\Feed\Outcome; + +$consumer->consume(function (CloudEvent $event) use ($mailer): ?Outcome { + if ($event->data['address'] === null) { + return Outcome::Skip; // malformed forever — stepping over it is the decision + } + + if (!$mailer->healthy()) { + return Outcome::Retry; // known-down dependency — same event next run + } + + $mailer->send($event->data); + + return null; // Continue +}); +``` + +### Consuming in chunks + +`consumeChunk()` is `consume()` with the whole poll — up to `batch` events — +handed over as one `list`, for handlers whose work is cheaper in +bulk: a multi-row upsert, one pipeline instead of a call per event. The +outcome vocabulary is the same, but the answer covers the chunk: the position +moves past all of it or none of it, so a retried chunk is re-delivered whole +and an idempotent handler absorbs the overlap. The handler is not called for +an empty poll. + +```php +$consumer = new Consumer($client, $cursor, name: 'projector', feed: 'edge', batch: 500); + +$consumer->consumeChunk(function (array $events) use ($db): ?Outcome { + try { + $db->upsertMany(\array_map(fn (CloudEvent $event) => $event->data, $events)); + } catch (DeadlockException) { + return Outcome::Retry; // transient — the same chunk comes back next run + } + + return null; +}); +``` + +A chunk that failed midway does not have to give its progress back: +`seek()` to the last event that succeeded before returning `Retry`, and the +next run starts strictly after it. A move made mid-run is never saved over, +so the chunk's own end does not overwrite the seek on the way out. + ### Starting at the tip A consumer with no stored position starts at the oldest retained event. A @@ -226,11 +292,13 @@ be arranged away: Every one re-delivers; none skips. An idempotent handler absorbs a duplicate, whereas an event stepped over is gone. -**Reject by throwing.** The run stops there, the position stays before the +**Reject by throwing** (or by returning `Outcome::Retry` — the same decision, +without an exception). The run stops there, the position stays before the failed event, and the next run retries it. Everything handled earlier in the run stays handled. A handler that keeps failing blocks everything behind it — intentionally: a feed is ordered, and stepping over a failure would apply -later events on top of state that was never updated. +later events on top of state that was never updated. Stepping over is never +implied; it is said explicitly, as `Outcome::Skip` or a `seek()`. **No position means the oldest retained event, never the tip** (unless the consumer opted into `Consumer::START_TIP`), so a consumer deployed after the producer diff --git a/src/Feed/Consumer.php b/src/Feed/Consumer.php index f09d5f9..e292028 100644 --- a/src/Feed/Consumer.php +++ b/src/Feed/Consumer.php @@ -5,6 +5,7 @@ namespace Utopia\Feed; use Utopia\Client\Adapter; +use Utopia\CloudEvents\CloudEvent; class Consumer { @@ -59,19 +60,19 @@ public function getName(): string return $this->name; } + /** + * The handler receives one event at a time and answers with an + * {@see Outcome}; returning anything else (or nothing) is + * {@see Outcome::Continue}, and throwing stops the run like + * {@see Outcome::Retry} with the error re-raised. + * + * @param callable(CloudEvent): mixed $handler + * @return int How many events the position advanced past — skipped ones included. + */ public function consume(callable $handler): int { $moved = $this->moved; - - $events = $this->feed->poll( - $this->position() ?? $this->origin(), - \max(1, \min($this->batch, Readable::MAX_BATCH)), - \max(0, \min($this->timeout, Readable::MAX_TIMEOUT)), - ); - - if ($events === []) { - return 0; - } + $events = $this->poll(); $handled = 0; $processed = null; @@ -79,29 +80,21 @@ public function consume(callable $handler): int foreach ($events as $event) { try { - $handler($event); + $outcome = $handler($event); } catch (\Throwable $error) { $failure = $error; break; } + if ($outcome === Outcome::Retry) { + break; + } + $processed = $event->id; $handled++; } - if ($processed !== null && $this->moved === $moved) { - $expected = $this->position; - $this->position = $processed; - - // Conditional for the same reason as the $moved guard, but across - // instances: a save lands only if the position is still where this - // run started. Refused means another instance moved it — progress, - // a seek, or a reset — and that newer decision stands. - if (!$this->cursor->advance($this->feed->getName(), $this->name, $processed, $expected)) { - $this->position = null; - $this->restored = false; - } - } + $this->advance($processed, $moved); if ($failure !== null) { throw $failure; @@ -110,6 +103,67 @@ public function consume(callable $handler): int return $handled; } + /** + * Like {@see Consumer::consume()}, but the handler receives the whole + * poll — up to `batch` events — as one `list`, and its + * {@see Outcome} answers for all of them: the position moves past the + * chunk or not at all. A handler that made partial progress before + * deciding {@see Outcome::Retry} can {@see Consumer::seek()} to the last + * event it completed; a move made mid-run is never saved over. + * + * The handler is not called for an empty poll — a caught-up consumer has + * nothing to decide about. + * + * @param callable(list): mixed $handler + * @return int How many events the position advanced past — the chunk, or 0. + */ + public function consumeChunk(callable $handler): int + { + $moved = $this->moved; + $events = $this->poll(); + + if ($events === []) { + return 0; + } + + if ($handler($events) === Outcome::Retry) { + return 0; + } + + $this->advance($events[\array_key_last($events)]->id, $moved); + + return \count($events); + } + + /** @return list */ + private function poll(): array + { + return $this->feed->poll( + $this->position() ?? $this->origin(), + \max(1, \min($this->batch, Readable::MAX_BATCH)), + \max(0, \min($this->timeout, Readable::MAX_TIMEOUT)), + ); + } + + private function advance(?string $processed, int $moved): void + { + if ($processed === null || $this->moved !== $moved) { + return; + } + + $expected = $this->position; + $this->position = $processed; + + // Conditional for the same reason as the $moved guard, but across + // instances: a save lands only if the position is still where this + // run started. Refused means another instance moved it — progress, + // a seek, or a reset — and that newer decision stands. + if (!$this->cursor->advance($this->feed->getName(), $this->name, $processed, $expected)) { + $this->position = null; + $this->restored = false; + } + } + private function origin(): ?string { return $this->start === self::START_TIP ? Readable::TIP : null; diff --git a/src/Feed/Outcome.php b/src/Feed/Outcome.php new file mode 100644 index 0000000..bdb0cc7 --- /dev/null +++ b/src/Feed/Outcome.php @@ -0,0 +1,34 @@ +assertSame(2, $attempts, 'The failed event is retried, not dropped'); } + /** + * Skip is the in-band form of the seek() escape hatch: the handler has + * already decided the event cannot be processed, so it steps past it + * without stopping the run — and a stepped-over event is gone. + */ + public function testSkipStepsPastOneEventWithoutStoppingTheRun(): void + { + $this->producer->produce('a'); + $this->producer->produce('poison'); + $last = $this->producer->produce('c'); + + $consumer = $this->consumer(); + $seen = []; + + $count = $consumer->consume(function (CloudEvent $event) use (&$seen): ?Outcome { + if ($event->type === 'poison') { + return Outcome::Skip; + } + + $seen[] = $event->type; + + return null; + }); + + $this->assertSame(3, $count, 'A skipped event still advances the position past it'); + $this->assertSame(['a', 'c'], $seen); + $this->assertSame($last, $this->cursor->load($this->name, 'invalidator')); + $this->assertSame([], $this->drain($consumer), 'The skipped event does not come back'); + } + + /** Retry is throwing without the exception: same position, calm return. */ + public function testRetryStopsTheRunAndKeepsTheProgressBeforeIt(): void + { + $first = $this->producer->produce('a'); + $this->producer->produce('b'); + $this->producer->produce('c'); + + $consumer = $this->consumer(); + + $count = $consumer->consume(fn (CloudEvent $event): ?Outcome => $event->type === 'b' ? Outcome::Retry : null); + + $this->assertSame(1, $count, 'Only what came before the retry is committed'); + $this->assertSame($first, $this->cursor->load($this->name, 'invalidator'), 'Progress before the retry is committed'); + $this->assertSame(['b', 'c'], $this->drain($consumer), 'The retried event comes back first, nothing behind it is lost'); + } + public function testDrainsABacklogInBatches(): void { foreach (\range(1, 10) as $i) { @@ -292,6 +339,173 @@ public function testDrainsABacklogInBatches(): void $this->assertSame(0, $consumer->consume(fn (CloudEvent $event) => null)); } + /** + * The chunk's event types, for asserting delivery. + * + * @param list $events + * @return list + */ + private static function types(array $events): array + { + return \array_map(static fn (CloudEvent $event): string => $event->type, $events); + } + + public function testAChunkHandlerReceivesTheWholePollAndAdvancesPastIt(): void + { + $this->producer->produce('a'); + $last = $this->producer->produce('b'); + + $consumer = $this->consumer(); + $chunks = []; + + $count = $consumer->consumeChunk(function (array $events) use (&$chunks): void { + $chunks[] = self::types($events); + }); + + $this->assertSame(2, $count); + $this->assertSame([['a', 'b']], $chunks, 'One call, the whole poll'); + $this->assertSame($last, $this->cursor->load($this->name, 'invalidator')); + $this->assertSame($last, $consumer->position()); + } + + public function testChunksFollowTheBatchSetting(): void + { + foreach (\range(1, 10) as $i) { + $this->producer->produce('event-' . $i); + } + + $consumer = $this->consumer(batch: 4); + $sizes = []; + $handler = function (array $events) use (&$sizes): void { + $sizes[] = \count($events); + }; + + $this->assertSame(4, $consumer->consumeChunk($handler)); + $this->assertSame(4, $consumer->consumeChunk($handler)); + $this->assertSame(2, $consumer->consumeChunk($handler)); + $this->assertSame(0, $consumer->consumeChunk($handler)); + + $this->assertSame([4, 4, 2], $sizes, 'A caught-up poll never reaches the handler'); + } + + public function testAChunkHandlerThatThrowsLeavesThePositionAlone(): void + { + $this->producer->produce('a'); + $this->producer->produce('b'); + + $consumer = $this->consumer(); + + try { + $consumer->consumeChunk(fn (array $events) => throw new \RuntimeException('nope')); + $this->fail('The handler failure should have been re-raised'); + } catch (\RuntimeException $error) { + $this->assertSame('nope', $error->getMessage()); + } + + $this->assertNull($this->cursor->load($this->name, 'invalidator'), 'A failed chunk commits nothing'); + + $seen = []; + $consumer->consumeChunk(function (array $events) use (&$seen): void { + $seen = self::types($events); + }); + + $this->assertSame(['a', 'b'], $seen, 'The whole chunk comes back'); + } + + public function testRetryRedeliversTheWholeChunkOnTheNextRun(): void + { + $this->producer->produce('a'); + $this->producer->produce('b'); + + $consumer = $this->consumer(); + + $this->assertSame(0, $consumer->consumeChunk(fn (array $events): Outcome => Outcome::Retry)); + $this->assertNull($this->cursor->load($this->name, 'invalidator'), 'Retry commits nothing'); + + $seen = []; + $count = $consumer->consumeChunk(function (array $events) use (&$seen): void { + $seen = self::types($events); + }); + + $this->assertSame(2, $count); + $this->assertSame(['a', 'b'], $seen); + } + + public function testSkipAdvancesPastAChunkItCouldNotProcess(): void + { + $this->producer->produce('a'); + $last = $this->producer->produce('b'); + + $consumer = $this->consumer(); + + $this->assertSame(2, $consumer->consumeChunk(fn (array $events): Outcome => Outcome::Skip)); + $this->assertSame($last, $this->cursor->load($this->name, 'invalidator'), 'A skipped chunk is stepped over, not retried'); + + $this->producer->produce('c'); + + $seen = []; + $consumer->consumeChunk(function (array $events) use (&$seen): void { + $seen = self::types($events); + }); + + $this->assertSame(['c'], $seen, 'The next run starts after the skipped chunk'); + } + + /** + * The partial-progress pattern: a chunk that failed midway seeks to the + * last event it completed and asks for a retry, so only the remainder + * comes back. The seek is the newer decision — the run must not save the + * chunk's end over it. + */ + public function testASeekMadeInsideAChunkHandlerIsNotOverwritten(): void + { + $this->producer->produce('a'); + $second = $this->producer->produce('b'); + $this->producer->produce('c'); + + $consumer = $this->consumer(); + + $consumer->consumeChunk(function (array $events) use ($consumer, $second): Outcome { + $consumer->seek($second); + + return Outcome::Retry; + }); + + $this->assertSame($second, $consumer->position()); + $this->assertSame($second, $this->cursor->load($this->name, 'invalidator')); + + $seen = []; + $consumer->consumeChunk(function (array $events) use (&$seen): void { + $seen = self::types($events); + }); + + $this->assertSame(['c'], $seen, 'The run resumes after the seeked id, not before the chunk'); + } + + /** The moved guard, on the path where the chunk does try to save. */ + public function testAChunkRunDoesNotSaveOverASeekEvenWhenItContinues(): void + { + $this->producer->produce('a'); + $second = $this->producer->produce('b'); + $this->producer->produce('c'); + + $consumer = $this->consumer(); + + $consumer->consumeChunk(function (array $events) use ($consumer, $second): void { + $consumer->seek($second); + }); + + $this->assertSame($second, $consumer->position(), 'The seek stands over the chunk\'s own end'); + $this->assertSame($second, $this->cursor->load($this->name, 'invalidator')); + + $seen = []; + $consumer->consumeChunk(function (array $events) use (&$seen): void { + $seen = self::types($events); + }); + + $this->assertSame(['c'], $seen); + } + public function testTipStartDoesNotAnnounceTheBacklog(): void { $this->producer->produce('old-1'); diff --git a/tests/Feed/Unit/ConsumerTest.php b/tests/Feed/Unit/ConsumerTest.php index 7ba4cb7..877eedb 100644 --- a/tests/Feed/Unit/ConsumerTest.php +++ b/tests/Feed/Unit/ConsumerTest.php @@ -239,6 +239,58 @@ public function testAPositionThatCannotBeSavedIsRaisedAfterTheEventsAreHandled() $this->assertNotNull($consumer->position(), 'The in-memory position still moved'); } + /** + * The chunk was handled, so the save failure comes after it — the same + * promise consume() makes, on the chunk path: this run keeps its progress + * in memory and only a restart replays. + */ + public function testAChunkPositionThatCannotBeSavedIsRaisedAfterTheChunkWasHandled(): void + { + $this->producer->produce('a'); + $this->producer->produce('b'); + + $consumer = $this->consumer(new FailingCursor(onSave: true)); + $seen = []; + + try { + $consumer->consumeChunk(function (array $events) use (&$seen): void { + $seen = \array_map(static fn (CloudEvent $event): string => $event->type, $events); + }); + $this->fail('The store failure should have been raised'); + } catch (Transport $error) { + $this->assertSame('Cursor store is unavailable', $error->getMessage()); + } + + $this->assertSame(['a', 'b'], $seen, 'The handler still saw the chunk'); + $this->assertNotNull($consumer->position(), 'The in-memory position still moved'); + } + + /** The conditional save guards the chunk path too: a stale chunk run concedes. */ + public function testAStaleChunkRunCannotUndoAnotherInstancesProgress(): void + { + $one = $this->consumer(); + $two = $this->consumer(); + + // An empty first run restores "no position yet" on both instances... + $this->assertSame(0, $one->consumeChunk(fn (array $events) => null)); + $this->assertSame(0, $two->consumeChunk(fn (array $events) => null)); + + foreach (['a', 'b', 'c', 'd'] as $type) { + $this->producer->produce($type); + } + + // ...then the first instance gets ahead. + $one->consumeChunk(fn (array $events) => null); + $ahead = $one->position(); + $this->assertNotNull($ahead); + + // The stale instance re-handles the chunk (at-least-once), but its + // save is refused rather than moving the shared position back. + $two->consumeChunk(fn (array $events) => null); + $this->assertSame($ahead, $this->cursor->load('edge', 'invalidator')); + $this->assertSame($ahead, $two->position(), 'Having conceded, it adopts the shared position'); + } + /** * A seek that did not persist must not look like one that did: the store * failure surfaces, and the in-memory position stays where it was. From 42950cfaadcdb7e94899497f5ef02004cce7ad59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matej=20Ba=C4=8Do?= Date: Wed, 12 Aug 2026 13:17:20 +0200 Subject: [PATCH 2/5] =?UTF-8?q?Remove=20Skip=20=E2=80=94=20it=20was=20Cont?= =?UTF-8?q?inue=20wearing=20a=20costume?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mechanically Skip and Continue were the same path: advance, keep going. An enum case with no branch behind it is API without function, so it goes; stepping over an event remains a seek(), said explicitly. Retry stays an enum rather than becoming a boolean: PHP APIs return false all the time, and a handler whose last statement happens to return one must not stall the feed by accident. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 12 ++++----- README.md | 13 ++++------ src/Feed/Consumer.php | 2 +- src/Feed/Outcome.php | 14 +++++----- tests/Feed/Consumer/Base.php | 50 ------------------------------------ 5 files changed, 18 insertions(+), 73 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fcbd2eb..47a68b9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,12 +4,12 @@ ### Consuming -- `Outcome` — the words a handler answers with: `Continue` (processed, advance), - `Skip` (cannot be processed and will not be retried — advance anyway, said - explicitly or not at all), and `Retry` (stop here, same position, so the next - run re-delivers — throwing with the exception left out). A handler that - returns anything else, or nothing, has continued, so every existing handler - keeps its meaning. +- `Outcome` — the words a handler answers with: `Continue` (processed, advance) + and `Retry` (stop here, same position, so the next run re-delivers — throwing + with the exception left out). A handler that returns anything else, or + nothing, has continued, so every existing handler keeps its meaning — and an + accidental `false` from the handler's last statement cannot stall the feed, + which is why this is an enum and not a boolean. - `Consumer::consumeChunk(callable)` — the whole poll (up to `batch` events) as one `list` per call, for handlers whose work is cheaper in bulk. One outcome answers for the chunk: the position moves past all of it or none diff --git a/README.md b/README.md index 5f811eb..a27b2b1 100644 --- a/README.md +++ b/README.md @@ -168,24 +168,21 @@ saying "processed, move on": | The handler | The position | The run | | --- | --- | --- | | returns `Outcome::Continue` (or anything else) | advances past the event | keeps going | -| returns `Outcome::Skip` | advances past the event | keeps going | | returns `Outcome::Retry` | stays before the event | ends, returns the count so far | | throws | stays before the event | ends, the error is re-raised | `Retry` and throwing are the same decision about the feed — this event is not handled, re-deliver it — made in different moods: throwing is for accidents and surfaces the error, `Retry` is for failures the handler expected (a -dependency it already knows is down) and returns calmly. `Skip` is the -deliberate loss of an event, decided in code: +dependency it already knows is down) and returns calmly. It is an enum rather +than a boolean on purpose: PHP APIs return `false` all the time, and a handler +whose last statement happens to return one must not stall the feed by +accident — retrying can only be said deliberately. ```php use Utopia\Feed\Outcome; $consumer->consume(function (CloudEvent $event) use ($mailer): ?Outcome { - if ($event->data['address'] === null) { - return Outcome::Skip; // malformed forever — stepping over it is the decision - } - if (!$mailer->healthy()) { return Outcome::Retry; // known-down dependency — same event next run } @@ -298,7 +295,7 @@ failed event, and the next run retries it. Everything handled earlier in the run stays handled. A handler that keeps failing blocks everything behind it — intentionally: a feed is ordered, and stepping over a failure would apply later events on top of state that was never updated. Stepping over is never -implied; it is said explicitly, as `Outcome::Skip` or a `seek()`. +implied; it is a `seek()`, said explicitly. **No position means the oldest retained event, never the tip** (unless the consumer opted into `Consumer::START_TIP`), so a consumer deployed after the producer diff --git a/src/Feed/Consumer.php b/src/Feed/Consumer.php index e292028..9acf56e 100644 --- a/src/Feed/Consumer.php +++ b/src/Feed/Consumer.php @@ -67,7 +67,7 @@ public function getName(): string * {@see Outcome::Retry} with the error re-raised. * * @param callable(CloudEvent): mixed $handler - * @return int How many events the position advanced past — skipped ones included. + * @return int How many events the position advanced past. */ public function consume(callable $handler): int { diff --git a/src/Feed/Outcome.php b/src/Feed/Outcome.php index bdb0cc7..6b985aa 100644 --- a/src/Feed/Outcome.php +++ b/src/Feed/Outcome.php @@ -10,21 +10,19 @@ * * Returning nothing decides too: a handler that returns normally without an * Outcome has continued, so every handler written before this enum existed - * keeps its meaning. Throwing is the fourth word in the vocabulary — the + * keeps its meaning. Throwing is the third word in the vocabulary — the * position stays, like Retry, and the error reaches the caller. + * + * Deliberately an enum rather than a boolean: PHP APIs return false all the + * time, so a handler whose last statement happens to return one must not + * acquire retry semantics by accident — a persistent false would stall the + * feed silently. Retry can only be said on purpose. */ enum Outcome { /** Processed — advance the position past it. */ case Continue; - /** - * Could not be processed, and retrying will not change that — advance - * anyway. The decision to lose an event is the handler's to make, so it - * is never implied: only this explicit word steps over a failure. - */ - case Skip; - /** * Something is wrong beyond this event — stop the run here. The position * stays before it, so the next run re-delivers it: the deliberate form of diff --git a/tests/Feed/Consumer/Base.php b/tests/Feed/Consumer/Base.php index 84d9191..fdd1d06 100644 --- a/tests/Feed/Consumer/Base.php +++ b/tests/Feed/Consumer/Base.php @@ -279,36 +279,6 @@ public function testAFailedEventBlocksTheOnesBehindItUntilItSucceeds(): void $this->assertSame(2, $attempts, 'The failed event is retried, not dropped'); } - /** - * Skip is the in-band form of the seek() escape hatch: the handler has - * already decided the event cannot be processed, so it steps past it - * without stopping the run — and a stepped-over event is gone. - */ - public function testSkipStepsPastOneEventWithoutStoppingTheRun(): void - { - $this->producer->produce('a'); - $this->producer->produce('poison'); - $last = $this->producer->produce('c'); - - $consumer = $this->consumer(); - $seen = []; - - $count = $consumer->consume(function (CloudEvent $event) use (&$seen): ?Outcome { - if ($event->type === 'poison') { - return Outcome::Skip; - } - - $seen[] = $event->type; - - return null; - }); - - $this->assertSame(3, $count, 'A skipped event still advances the position past it'); - $this->assertSame(['a', 'c'], $seen); - $this->assertSame($last, $this->cursor->load($this->name, 'invalidator')); - $this->assertSame([], $this->drain($consumer), 'The skipped event does not come back'); - } - /** Retry is throwing without the exception: same position, calm return. */ public function testRetryStopsTheRunAndKeepsTheProgressBeforeIt(): void { @@ -431,26 +401,6 @@ public function testRetryRedeliversTheWholeChunkOnTheNextRun(): void $this->assertSame(['a', 'b'], $seen); } - public function testSkipAdvancesPastAChunkItCouldNotProcess(): void - { - $this->producer->produce('a'); - $last = $this->producer->produce('b'); - - $consumer = $this->consumer(); - - $this->assertSame(2, $consumer->consumeChunk(fn (array $events): Outcome => Outcome::Skip)); - $this->assertSame($last, $this->cursor->load($this->name, 'invalidator'), 'A skipped chunk is stepped over, not retried'); - - $this->producer->produce('c'); - - $seen = []; - $consumer->consumeChunk(function (array $events) use (&$seen): void { - $seen = self::types($events); - }); - - $this->assertSame(['c'], $seen, 'The next run starts after the skipped chunk'); - } - /** * The partial-progress pattern: a chunk that failed midway seeks to the * last event it completed and asks for a retry, so only the remainder From c4829e3e0a2eb119e9665ddcd24e58cf612c5d03 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matej=20Ba=C4=8Do?= Date: Wed, 12 Aug 2026 13:25:40 +0200 Subject: [PATCH 3/5] Replace the Outcome enum with a boolean answer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A handler now answers with its return value: an exact false means unprocessed — same position, next run re-delivers, throwing with the exception left out. Anything else, including nothing, means processed, so every existing handler keeps its meaning: only that exact false counts, never a stray null, 0 or ''. The trade this makes is documented where it bites: PHP APIs answer false to mean failure, so a handler ending in `return $api->call()` says "retry until it succeeds" — write the return you mean. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 19 ++++++------ README.md | 56 +++++++++++++++++++----------------- src/Feed/Consumer.php | 25 +++++++++------- src/Feed/Outcome.php | 32 --------------------- tests/Feed/Consumer/Base.php | 44 +++++++++++++++++++--------- 5 files changed, 83 insertions(+), 93 deletions(-) delete mode 100644 src/Feed/Outcome.php diff --git a/CHANGELOG.md b/CHANGELOG.md index 47a68b9..4742407 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,18 +4,17 @@ ### Consuming -- `Outcome` — the words a handler answers with: `Continue` (processed, advance) - and `Retry` (stop here, same position, so the next run re-delivers — throwing - with the exception left out). A handler that returns anything else, or - nothing, has continued, so every existing handler keeps its meaning — and an - accidental `false` from the handler's last statement cannot stall the feed, - which is why this is an enum and not a boolean. +- A handler's return value now answers for the event: an exact `false` means + unprocessed — the run stops there, same position, so the next run + re-delivers it (throwing with the exception left out). Anything else, + including nothing, means processed, so every existing handler keeps its + meaning: only that exact `false` counts, never a stray `null`, `0` or `''`. - `Consumer::consumeChunk(callable)` — the whole poll (up to `batch` events) as one `list` per call, for handlers whose work is cheaper in bulk. - One outcome answers for the chunk: the position moves past all of it or none - of it. A handler that made partial progress can `seek()` to the last event it - completed before returning `Retry` — a move made mid-run is never saved over. - The handler is not called for an empty poll. + One answer covers the chunk: the position moves past all of it or, on + `false`, none of it. A handler that made partial progress can `seek()` to the + last event it completed before returning `false` — a move made mid-run is + never saved over. The handler is not called for an empty poll. ## 0.1.0 — Initial release diff --git a/README.md b/README.md index a27b2b1..fb12295 100644 --- a/README.md +++ b/README.md @@ -159,66 +159,68 @@ names its feed: $consumer = new Consumer($store, new Cursor\Redis($redis), name: 'audit-log'); ``` -### Answering with an outcome +### Answering with the return value -A handler answers with an `Outcome`; anything else it returns — including -nothing — reads as `Outcome::Continue`, so a handler that simply returns is -saying "processed, move on": +A handler's return value answers for the event: an exact `false` means +"unprocessed" — the run stops there, the position stays before the event, and +the next run re-delivers it. Anything else, including nothing, means +"processed, move on", so a handler that simply returns keeps its meaning: | The handler | The position | The run | | --- | --- | --- | -| returns `Outcome::Continue` (or anything else) | advances past the event | keeps going | -| returns `Outcome::Retry` | stays before the event | ends, returns the count so far | +| returns `false` — exactly | stays before the event | ends, returns the count so far | +| returns anything else, or nothing | advances past the event | keeps going | | throws | stays before the event | ends, the error is re-raised | -`Retry` and throwing are the same decision about the feed — this event is not +`false` and throwing are the same decision about the feed — this event is not handled, re-deliver it — made in different moods: throwing is for accidents -and surfaces the error, `Retry` is for failures the handler expected (a -dependency it already knows is down) and returns calmly. It is an enum rather -than a boolean on purpose: PHP APIs return `false` all the time, and a handler -whose last statement happens to return one must not stall the feed by -accident — retrying can only be said deliberately. +and surfaces the error, `false` is for failures the handler expected (a +dependency it already knows is down) and returns calmly. ```php -use Utopia\Feed\Outcome; - -$consumer->consume(function (CloudEvent $event) use ($mailer): ?Outcome { +$consumer->consume(function (CloudEvent $event) use ($mailer): bool { if (!$mailer->healthy()) { - return Outcome::Retry; // known-down dependency — same event next run + return false; // known-down dependency — same event next run } $mailer->send($event->data); - return null; // Continue + return true; }); ``` +Only that exact `false` counts — `null`, `0` and `''` all mean processed — so +no handler written before this contract can stall the feed. The flip side +deserves a moment's care: PHP APIs answer `false` to mean failure, so a +handler ending in `return $mailer->send($event->data);` says "retry until it +sends". Write the `return` you mean. + ### Consuming in chunks `consumeChunk()` is `consume()` with the whole poll — up to `batch` events — handed over as one `list`, for handlers whose work is cheaper in bulk: a multi-row upsert, one pipeline instead of a call per event. The -outcome vocabulary is the same, but the answer covers the chunk: the position -moves past all of it or none of it, so a retried chunk is re-delivered whole -and an idempotent handler absorbs the overlap. The handler is not called for -an empty poll. +return value is read the same way, but the answer covers the chunk: the +position moves past all of it or, on `false`, none of it, so an unprocessed +chunk is re-delivered whole and an idempotent handler absorbs the overlap. +The handler is not called for an empty poll. ```php $consumer = new Consumer($client, $cursor, name: 'projector', feed: 'edge', batch: 500); -$consumer->consumeChunk(function (array $events) use ($db): ?Outcome { +$consumer->consumeChunk(function (array $events) use ($db): bool { try { $db->upsertMany(\array_map(fn (CloudEvent $event) => $event->data, $events)); } catch (DeadlockException) { - return Outcome::Retry; // transient — the same chunk comes back next run + return false; // transient — the same chunk comes back next run } - return null; + return true; }); ``` A chunk that failed midway does not have to give its progress back: -`seek()` to the last event that succeeded before returning `Retry`, and the +`seek()` to the last event that succeeded before returning `false`, and the next run starts strictly after it. A move made mid-run is never saved over, so the chunk's own end does not overwrite the seek on the way out. @@ -289,8 +291,8 @@ be arranged away: Every one re-delivers; none skips. An idempotent handler absorbs a duplicate, whereas an event stepped over is gone. -**Reject by throwing** (or by returning `Outcome::Retry` — the same decision, -without an exception). The run stops there, the position stays before the +**Reject by throwing** (or by returning `false` — the same decision, without +an exception). The run stops there, the position stays before the failed event, and the next run retries it. Everything handled earlier in the run stays handled. A handler that keeps failing blocks everything behind it — intentionally: a feed is ordered, and stepping over a failure would apply diff --git a/src/Feed/Consumer.php b/src/Feed/Consumer.php index 9acf56e..28e5ec7 100644 --- a/src/Feed/Consumer.php +++ b/src/Feed/Consumer.php @@ -61,10 +61,13 @@ public function getName(): string } /** - * The handler receives one event at a time and answers with an - * {@see Outcome}; returning anything else (or nothing) is - * {@see Outcome::Continue}, and throwing stops the run like - * {@see Outcome::Retry} with the error re-raised. + * The handler receives one event at a time and answers with its return + * value: an exact `false` means unprocessed — the run stops there, the + * position stays before the event, and the next run re-delivers it, the + * calm form of what throwing does. Anything else, including nothing, + * means processed. Only that exact `false` counts: handlers predate this + * contract and return all sorts of things, and a stray null or 0 must + * not stall the feed. * * @param callable(CloudEvent): mixed $handler * @return int How many events the position advanced past. @@ -80,13 +83,13 @@ public function consume(callable $handler): int foreach ($events as $event) { try { - $outcome = $handler($event); + $result = $handler($event); } catch (\Throwable $error) { $failure = $error; break; } - if ($outcome === Outcome::Retry) { + if ($result === false) { break; } @@ -106,10 +109,10 @@ public function consume(callable $handler): int /** * Like {@see Consumer::consume()}, but the handler receives the whole * poll — up to `batch` events — as one `list`, and its - * {@see Outcome} answers for all of them: the position moves past the - * chunk or not at all. A handler that made partial progress before - * deciding {@see Outcome::Retry} can {@see Consumer::seek()} to the last - * event it completed; a move made mid-run is never saved over. + * return value answers for all of them: the position moves past the + * chunk or, on an exact `false`, not at all. A handler that made partial + * progress before answering `false` can {@see Consumer::seek()} to the + * last event it completed; a move made mid-run is never saved over. * * The handler is not called for an empty poll — a caught-up consumer has * nothing to decide about. @@ -126,7 +129,7 @@ public function consumeChunk(callable $handler): int return 0; } - if ($handler($events) === Outcome::Retry) { + if ($handler($events) === false) { return 0; } diff --git a/src/Feed/Outcome.php b/src/Feed/Outcome.php deleted file mode 100644 index 6b985aa..0000000 --- a/src/Feed/Outcome.php +++ /dev/null @@ -1,32 +0,0 @@ -assertSame(2, $attempts, 'The failed event is retried, not dropped'); } - /** Retry is throwing without the exception: same position, calm return. */ - public function testRetryStopsTheRunAndKeepsTheProgressBeforeIt(): void + /** Returning false is throwing without the exception: same position, calm return. */ + public function testAFalseReturnStopsTheRunAndKeepsTheProgressBeforeIt(): void { $first = $this->producer->produce('a'); $this->producer->produce('b'); @@ -288,11 +287,30 @@ public function testRetryStopsTheRunAndKeepsTheProgressBeforeIt(): void $consumer = $this->consumer(); - $count = $consumer->consume(fn (CloudEvent $event): ?Outcome => $event->type === 'b' ? Outcome::Retry : null); + $count = $consumer->consume(fn (CloudEvent $event): bool => $event->type !== 'b'); - $this->assertSame(1, $count, 'Only what came before the retry is committed'); - $this->assertSame($first, $this->cursor->load($this->name, 'invalidator'), 'Progress before the retry is committed'); - $this->assertSame(['b', 'c'], $this->drain($consumer), 'The retried event comes back first, nothing behind it is lost'); + $this->assertSame(1, $count, 'Only what came before the false is committed'); + $this->assertSame($first, $this->cursor->load($this->name, 'invalidator'), 'Progress before the false is committed'); + $this->assertSame(['b', 'c'], $this->drain($consumer), 'The unprocessed event comes back first, nothing behind it is lost'); + } + + /** + * Only an exact false answers "unprocessed": handlers predate this + * contract and return all sorts of things — null, counts, empty strings — + * and none of them may stall the feed. + */ + public function testOnlyAnExactFalseStopsTheRun(): void + { + $this->producer->produce('a'); + $this->producer->produce('b'); + + $consumer = $this->consumer(); + + $this->assertSame(2, $consumer->consume(fn (CloudEvent $event) => 0), 'A falsy non-false still means processed'); + + $this->producer->produce('c'); + + $this->assertSame(1, $consumer->consumeChunk(fn (array $events) => ''), 'The chunk answer is judged the same way'); } public function testDrainsABacklogInBatches(): void @@ -382,15 +400,15 @@ public function testAChunkHandlerThatThrowsLeavesThePositionAlone(): void $this->assertSame(['a', 'b'], $seen, 'The whole chunk comes back'); } - public function testRetryRedeliversTheWholeChunkOnTheNextRun(): void + public function testAFalseReturnRedeliversTheWholeChunkOnTheNextRun(): void { $this->producer->produce('a'); $this->producer->produce('b'); $consumer = $this->consumer(); - $this->assertSame(0, $consumer->consumeChunk(fn (array $events): Outcome => Outcome::Retry)); - $this->assertNull($this->cursor->load($this->name, 'invalidator'), 'Retry commits nothing'); + $this->assertSame(0, $consumer->consumeChunk(fn (array $events): bool => false)); + $this->assertNull($this->cursor->load($this->name, 'invalidator'), 'A false answer commits nothing'); $seen = []; $count = $consumer->consumeChunk(function (array $events) use (&$seen): void { @@ -403,7 +421,7 @@ public function testRetryRedeliversTheWholeChunkOnTheNextRun(): void /** * The partial-progress pattern: a chunk that failed midway seeks to the - * last event it completed and asks for a retry, so only the remainder + * last event it completed and answers false, so only the remainder * comes back. The seek is the newer decision — the run must not save the * chunk's end over it. */ @@ -415,10 +433,10 @@ public function testASeekMadeInsideAChunkHandlerIsNotOverwritten(): void $consumer = $this->consumer(); - $consumer->consumeChunk(function (array $events) use ($consumer, $second): Outcome { + $consumer->consumeChunk(function (array $events) use ($consumer, $second): bool { $consumer->seek($second); - return Outcome::Retry; + return false; }); $this->assertSame($second, $consumer->position()); From b7abbccac335300a56d258fe843c0399d8d365b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matej=20Ba=C4=8Do?= Date: Wed, 12 Aug 2026 13:41:26 +0200 Subject: [PATCH 4/5] Document the count as handler progress, not cursor movement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The docblocks claimed the return counts events the position advanced past, but a mid-run seek()/reset() (the moved guard) or a refused conditional save (another instance got there first) leaves the position elsewhere while the count still describes what the handler did. The cursor is shared and movable by hand from any process, so advancement was never a contract the count could keep — say what it truthfully is, and pin the seek-mid-chunk return in the test that exercises it. Co-Authored-By: Claude Fable 5 --- src/Feed/Consumer.php | 11 +++++++++-- tests/Feed/Consumer/Base.php | 3 ++- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/src/Feed/Consumer.php b/src/Feed/Consumer.php index 28e5ec7..d6f5514 100644 --- a/src/Feed/Consumer.php +++ b/src/Feed/Consumer.php @@ -70,7 +70,12 @@ public function getName(): string * not stall the feed. * * @param callable(CloudEvent): mixed $handler - * @return int How many events the position advanced past. + * @return int How many events the handler processed this run. The + * position follows it past the last of them — unless it was + * moved by hand mid-run ({@see Consumer::seek()}, + * {@see Consumer::reset()}) or another instance moved it + * first, in which case that newer decision stands and this + * count does not describe it. */ public function consume(callable $handler): int { @@ -118,7 +123,9 @@ public function consume(callable $handler): int * nothing to decide about. * * @param callable(list): mixed $handler - * @return int How many events the position advanced past — the chunk, or 0. + * @return int How many events the handler processed — the chunk, or 0 on + * `false`. As in {@see Consumer::consume()}, a position moved + * by hand mid-run is not described by this count. */ public function consumeChunk(callable $handler): int { diff --git a/tests/Feed/Consumer/Base.php b/tests/Feed/Consumer/Base.php index eed499a..7a6e4da 100644 --- a/tests/Feed/Consumer/Base.php +++ b/tests/Feed/Consumer/Base.php @@ -459,10 +459,11 @@ public function testAChunkRunDoesNotSaveOverASeekEvenWhenItContinues(): void $consumer = $this->consumer(); - $consumer->consumeChunk(function (array $events) use ($consumer, $second): void { + $count = $consumer->consumeChunk(function (array $events) use ($consumer, $second): void { $consumer->seek($second); }); + $this->assertSame(3, $count, 'The count is the handler\'s progress; the seek owns the position'); $this->assertSame($second, $consumer->position(), 'The seek stands over the chunk\'s own end'); $this->assertSame($second, $this->cursor->load($this->name, 'invalidator')); From e8059f09ef1af0370d053ba17cd91b4576690222 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matej=20Ba=C4=8Do?= Date: Wed, 12 Aug 2026 13:41:58 +0200 Subject: [PATCH 5/5] Stamp the changelog as 0.1.1 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4742407..5a6261d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## Unreleased +## 0.1.1 — Chunk consumption and handler answers ### Consuming