Add handler outcomes and chunk consumption

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<CloudEvent>` 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

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<CloudEvent>`, 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

src/Feed/Consumer.php

@@ -5,6 +5,7 @@
 namespace Utopia\Feed;
 
 use Utopia\Client\Adapter;
+use Utopia\CloudEvents\CloudEvent;
 
 class Consumer
 {
@@ -59,49 +60,49 @@ 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;
         $failure = null;
 
         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<CloudEvent>`, 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<CloudEvent>): 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<CloudEvent> */
+    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;