diff --git a/CHANGELOG.md b/CHANGELOG.md index e6d7031..5a6261d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # Changelog +## 0.1.1 — Chunk consumption and handler answers + +### Consuming + +- 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 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 Pull-based HTTP event feeds ([http-feeds.org](https://www.http-feeds.org/)) for diff --git a/README.md b/README.md index 4230d14..fb12295 100644 --- a/README.md +++ b/README.md @@ -159,6 +159,71 @@ names its feed: $consumer = new Consumer($store, new Cursor\Redis($redis), name: 'audit-log'); ``` +### Answering with the return value + +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 `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 | + +`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, `false` is for failures the handler expected (a +dependency it already knows is down) and returns calmly. + +```php +$consumer->consume(function (CloudEvent $event) use ($mailer): bool { + if (!$mailer->healthy()) { + return false; // known-down dependency — same event next run + } + + $mailer->send($event->data); + + 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 +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): bool { + try { + $db->upsertMany(\array_map(fn (CloudEvent $event) => $event->data, $events)); + } catch (DeadlockException) { + return false; // transient — the same chunk comes back next run + } + + return true; +}); +``` + +A chunk that failed midway does not have to give its progress back: +`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. + ### Starting at the tip A consumer with no stored position starts at the oldest retained event. A @@ -226,11 +291,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 `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 -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 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 f09d5f9..d6f5514 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,27 @@ public function getName(): string return $this->name; } + /** + * 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 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 { $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 +88,21 @@ public function consume(callable $handler): int foreach ($events as $event) { try { - $handler($event); + $result = $handler($event); } catch (\Throwable $error) { $failure = $error; break; } + if ($result === false) { + 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 +111,69 @@ 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 + * 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. + * + * @param callable(list): mixed $handler + * @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 + { + $moved = $this->moved; + $events = $this->poll(); + + if ($events === []) { + return 0; + } + + if ($handler($events) === false) { + 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/tests/Feed/Consumer/Base.php b/tests/Feed/Consumer/Base.php index da5ab0d..7a6e4da 100644 --- a/tests/Feed/Consumer/Base.php +++ b/tests/Feed/Consumer/Base.php @@ -278,6 +278,41 @@ public function testAFailedEventBlocksTheOnesBehindItUntilItSucceeds(): void $this->assertSame(2, $attempts, 'The failed event is retried, not dropped'); } + /** Returning false is throwing without the exception: same position, calm return. */ + public function testAFalseReturnStopsTheRunAndKeepsTheProgressBeforeIt(): void + { + $first = $this->producer->produce('a'); + $this->producer->produce('b'); + $this->producer->produce('c'); + + $consumer = $this->consumer(); + + $count = $consumer->consume(fn (CloudEvent $event): bool => $event->type !== 'b'); + + $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 { foreach (\range(1, 10) as $i) { @@ -292,6 +327,154 @@ 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 testAFalseReturnRedeliversTheWholeChunkOnTheNextRun(): void + { + $this->producer->produce('a'); + $this->producer->produce('b'); + + $consumer = $this->consumer(); + + $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 { + $seen = self::types($events); + }); + + $this->assertSame(2, $count); + $this->assertSame(['a', 'b'], $seen); + } + + /** + * The partial-progress pattern: a chunk that failed midway seeks to the + * 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. + */ + 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): bool { + $consumer->seek($second); + + return false; + }); + + $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(); + + $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')); + + $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.