Feat: Initial implementation — pull-based HTTP event feeds - #1
Merged
Conversation
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Greptile SummaryThis PR introduces a pull-based HTTP event-feed library.
Confidence Score: 3/5The PR is not yet safe to merge because shared cursor updates can still overwrite newer progress, seeks, or resets. Cursor advancement compares the stored position and writes the replacement in separate operations, so another replica can update or delete the cursor between those operations and have its decision overwritten. Files Needing Attention: src/Feed/Cursor.php, src/Feed/Consumer.php, src/Feed/Cursor/Redis.php Important Files Changed
Reviews (18): Last reviewed commit: "Fix stability" | Re-trigger Greptile |
CI: appwrite/utopia-base:php-8.5-1.0.0 does not exist — 1.0.0 was only ever published for 8.3 and 8.4. Both the PHP 8.5 and PHPStan jobs failed at the FROM. Moved to 2.1.0, which is published for 8.3, 8.4 and 8.5. Protocol::decode() required the "events" field instead of defaulting it. A response with no envelope — a misrouted request, a proxy's JSON error page, an endpoint that moved — decoded to an empty batch, which a consumer reads as "caught up". That is indistinguishable from a genuinely empty feed and leaves the consumer parked at a position that never advances again. Journal now rejects a retention cap below one event. Non-positive caps meant opposite things per backend: Redis reads MAXLEN 0 as "keep nothing", while array_slice($events, -0) keeps everything, so the in-memory journal grew unbounded. Pinned actions/checkout to a commit SHA in all three workflows, so a moved upstream tag cannot change what CI executes. Documented that a consumer name belongs to one process: the position is written with a plain set, so two processes sharing a name can move it backwards. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two processes running the same consumer overlap during a rolling restart, which is a normal operation rather than a misconfiguration. They finish batches of different lengths, so the departing one's older position could land last and a later restart would replay everything between the two — on a busy feed, thousands of events. Positions are totally ordered, so Cursor::save() now compares before writing and drops anything that is not an advance. Cursor::shouldAdvance() holds the rule, and every implementation applies it. The guard fails open: a position that cannot be compared, because the store could not be read or because what came back is not a position, is treated as behind. It exists to stop a position going backwards and must never become a reason for one to stop going forwards. The comparison is not atomic. Closing the window entirely needs a compare-and-set in one operation, and each candidate costs more than the race does: Redis scores are doubles and cannot hold <ms>-<seq> exactly, WATCH leaves state on a connection about to return to a pool, and a one-entry stream changes the stored type and breaks cursors written by an earlier version. The window is microseconds against a poll interval of seconds, and losing the race costs a replay, which every handler already tolerates. Costs one extra read per advance. An idle poll saves nothing, so it still reads nothing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The read-compare-write in Cursor::save() is not atomic, so two processes sharing a consumer name can interleave and leave the older position stored. That was reasoned about but only thinly recorded, which leaves the next reader to redo the analysis. Consumer listed three reasons a handler must tolerate seeing an event twice. This is a fourth, and naming it there makes it part of the stated contract rather than an unlisted edge: every one of the four re-delivers, none of them skips, and that asymmetry is the point. Cursor::save() now carries why the window is left open — each compare-and-set candidate and what it costs, including that Cursor\Cache has no portable one at all, since a Utopia cache reports generation '0' on adapters without leases. A fix that held on some cache backends and silently not on others would be worse than one uniform, stated guarantee. Also drops a stale reference to a Lua script that no longer exists. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A feed position is a Redis stream id — <ms>-<seq> is the format XADD allocates — and Redis refuses to append an id equal to or smaller than the one at the top of a stream. Storing the position as the id of a one-entry stream makes "a position never moves backwards" the server's rule, enforced by the same operation that writes it, so two processes racing during a rolling restart cannot both decide they are ahead. It also drops a round trip: XADD replaces the read-compare-write. Cursor\Cache keeps the non-atomic guard. A Utopia cache has no portable compare-and-set — getGeneration() returns '0' on adapters without lease support — so a fix there would hold on some backends and silently not on others, which is worse than one clearly stated guarantee. Documented per implementation rather than as one blanket claim. The upgrade from string-valued keys is transparent: load() reads a string where it finds one, and the next save() replaces the key in place. The position is carried across, because the consumer is advancing past exactly what the string held, so nothing replays. docs/migration.md carries the operational notes — in particular that rolling back to a build predating this hits WRONGTYPE, and that anything reading these keys directly needs XREVRANGE rather than GET. Also drops the phpstan ignore rules for the Redis stream commands. They were added under phpstan 1.x; 2.2's bundled stubs cover these methods, so the rules matched nothing and, with reportUnmatchedIgnoredErrors off, would have silently masked real errors later. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
phpredis type() returns an integer constant, not a type name, so comparing it to 'stream' could never hold. And testCursorsAreStoredUnderTheFeedTheyBelongTo still read the key with GET, which returns false now the position lives in a stream entry's id. All three were assertions rather than behaviour: monotonicity, the digit boundary, the legacy read and the invalid-id rejection all passed against a real Redis on the previous run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A batch on the wire is now the plain JSON array of CloudEvents the spec
defines — the {total, events} envelope is gone. Protocol::encode()
returns the bare array, Protocol::decode() expects one (keeping the
per-entry leniency policy), and the HTTP journal asks for the spec's
application/cloudevents-batch+json media type while staying tolerant of
servers that answer application/json. The limit query parameter stays,
documented as an extension beyond the spec, and an event carrying the
spec's optional method attribute is pinned to decode into extensions.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Feed::read() and Feed::poll() now return a Batch — the events paired with the limit the read was actually clamped to, so cacheControl() can never be fed a number the batch was not built with. Batch counts and iterates as its events, toArray() is the wire encoding, and lastId() is the position a stateless relay tracks by hand. Feed::serve() takes a route's raw query-parameter array and does the whole request: extracts lastEventId, limit and timeout, coerces string values, applies the defaults and clamps, treats an empty lastEventId as absent, and rejects a malformed one with Exception\Invalid. A route body no longer names Protocol at all, and Protocol is documented as internal plumbing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A lastEventId of $ — a protocol extension like limit, borrowed from the Redis XREAD convention — now means "the tip of the feed". Journals resolve it to the newest entry at the moment of the call (an empty feed resolves to the beginning of future events), pinned once per poll so events landing mid-wait are still delivered. Journal\Http passes the sentinel through for the producer to resolve inside the same request, leaving no tip round trip to race; Feed::serve() lets it through its id validation, and Id keeps rejecting it. Consumers opt in with Start::Tip: when no position is stored, the first poll uses the sentinel. A stored cursor always wins, and reset() with Start::Tip means "forget everything, resume from now". Feed::tip() exposes the newest event's id for local journals. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
seek($eventId) treats the id as the last event handled: the next consume() starts strictly after it. The position persists immediately through the cursor store and mirrors in memory; a store failure surfaces as Transport with the in-memory position unmoved, so a seek that did not persist never looks like one that did. The id must satisfy Id::isValid — the tip sentinel included in the rejects — but does not need to exist in the feed, which is what makes seeking to a poison event's own id the deliberate way to step past it. Documented next to reset() in the README, with the poison-event recipe. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
The README promises `Exception\Transport` when "the backend or network failed: Redis errors, HTTP failures, a cursor store that is down", and `Store\Redis`/`Cursor\Redis` honour it by wrapping `\RedisException`. The cache-backed twins wrapped nothing. `Utopia\Cache\Adapter\Redis` rethrows `\RedisException` once its internal retries are exhausted, so over a Redis-backed cache that is down, `Store\Cache::read()/tip()/append()` and all three `Cursor\Cache` methods let a raw `\RedisException` escape. That breaks the base-class promise that every error this library raises extends `Utopia\Feed\Exception`, and it breaks the README's canonical consume loop: `catch (Transport)` does not catch it, so the consumer process dies on a backend blip instead of retrying — the exact failure mode the Transport contract exists to prevent. Both adapters now wrap the cache calls, following the Redis adapters' pattern. The try blocks stay narrow on purpose: `Cursor::key()` and the event decoding raise `Invalid`, which is the caller's bug rather than the backend's failure, and catching `\Throwable` around them would erase that distinction. A test pins it — an unusable name stays `Invalid` even when the backend behind the cursor is also down. `BrokenCache` grew a `raises` mode for this, so it now covers both ways a cache adapter fails: answering `false` and letting the backend's error out. It raises a plain exception rather than a `\RedisException` so the service-free suites stay service-free; what the wrapping cares about is only that the error is not one of ours. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Redis is one of several store adapters. `Store\Memory`, `Store\Cache` and consuming a feed over HTTP work without it, and the only references to `\Redis` are in type hints of classes a Redis-less service never instantiates. Declaring `ext-redis` under `require` claimed otherwise, so this package's own manifest overstated what its code needs. Moved to `suggest` with a note naming the four adapters that need it. One thing the review that prompted this assumed does not hold, and the README now says so plainly rather than implying the problem is solved: this change does not free a downstream install. `utopia-php/cache` — a hard dependency — requires `ext-redis` and `ext-memcached` itself, so a machine without them still needs `--ignore-platform-req` (verified against a scratch project: the resolver now stops at the cache package's extensions, not at ours). That constraint is not this repo's to remove. The README's development instructions traded the blanket `--ignore-platform-reqs` for three targeted `--ignore-platform-req=ext-*` flags. The blanket flag also skipped the PHP version check, which for a library requiring PHP 8.5 is the one platform requirement that must not be waived silently. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Feed and cursor keys were built by joining names with `:` — a character a name is allowed to contain, since names are only validated as non-empty. Joined raw, the mapping is not injective, and both collisions corrupt data rather than losing it: - A feed named `edge:cursor:x` keyed to `feed:edge:cursor:x`, which is also the cursor key of consumer `x` on feed `edge`. `Cursor\Redis::save()` then `SET`s over an `XADD` stream. The new Redis test shows this is not the harmless WRONGTYPE it looks like — the whole feed is gone afterwards. - Two distinct pairs could share one position: (`a:cursor:b`, `c`) and (`a`, `b:cursor:c`) both joined to `feed:a:cursor:b:cursor:c`, giving two unrelated consumers the "sharing a name" hazard the README warns about without anyone sharing a name. Percent-encoding `:`, and `%` itself so the encoding stays reversible, makes the collisions unexpressible. Escaping rather than rejecting is deliberate: `Remote` reads third-party feeds whose names are arbitrary path segments, and refusing one over a detail of how this library stores positions would make that feed unconsumable. A name containing neither character comes through untouched, so the layout the README documents — `feed:<name>`, `feed:<feed>:cursor:<consumer>` — still reads from a shell for every name anyone would actually pick. Both key shapes now live in one `Key` class. That also removes the duplicated `'feed:' . $this->name` in `Store\Redis` and `Store\Cache`, which was the reason a store could drift from the escaping the cursor does. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Most of what this task asked for already landed with the test rework: `Producer/PoolTest`, `Server/PoolTest` and `Consumer/PoolTest` run the shared scenarios over `Store\Pool` and `Cursor\Pool`, as their own CI job, so the delegates, their return values and the inner store's constructor validation are all exercised. One claim the review named was genuinely unpinned, and it is the one that matters most: `Store\Pool` exists because a long poll borrows per read rather than once around the whole loop, so a held poll never ties up a connection. Nothing tested that. Hoisting the borrow — an inviting "optimization", since it looks like it saves pool traffic — would have removed the class's entire reason to exist with the whole suite still green, and would only surface in production as a pool exhausted by idle consumers holding connections for up to 30 seconds each. `CountingStack` records releases so the property is asserted directly: a ~500ms poll on a 50ms interval must release many times, not once. Confirmed by hoisting the borrow — the test fails with "1 is greater than 2" and nothing else in the suite notices. Two leak tests come with it, on the store and cursor sides: after a round of reads, tips, seeks and resets the pool is whole again. A leaked connection fails nothing functionally until the pool runs dry, which in a service is minutes into production rather than here. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`Store\Cache` keeps the whole feed under one key, so `read()` fetched and re-validated every retained entry even to return nothing. Under the inherited poll loop that ran per `pollInterval` tick, per waiting consumer, for up to 30 seconds a request — and a caught-up consumer on a quiet feed is the common case, not the exception. The newest id now also lives under a small second key, so a poll that is provably caught up answers from that instead of loading the feed. Two properties keep it honest: - The marker is written *before* the feed, so it can be ahead but never behind. Behind, it would report a caught-up consumer and the event would never arrive; ahead only costs a read that finds nothing. - It is only ever used to skip a read, never to answer one. A missing, expired or unparseable marker falls straight through to the real read, and `tip()` still reads the feed rather than trusting it. Also lowered the adapter's default `maxSize` from the 100 000 it inherited from the Redis store to 1 000. That number was chosen for a backend where trimming is server-side and reads are ranged; here retention is also the size of every append's read-modify-write, so at the inherited default a single `produce()` moved megabytes through the cache in both directions. This is a behaviour change — a feed relying on 100 000 entries of retention now has to ask for it — which is why it is happening before the first release rather than after. Appends are still O(feed size): that is inherent to one key per feed, and the retention default is now scaled to it rather than hiding it. The README says all of this where the adapter is introduced. Costs are asserted, not just described: `CountingCache` records reads per key, and a 400ms caught-up poll must read the marker repeatedly and the feed exactly zero times. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
All three workflows triggered on `pull_request` only, so nothing ever ran on the default branch. The README's build badge points at `tests.yml`, a workflow with no runs on `main` — it would have shown "no status" indefinitely. More usefully, a merge that breaks `main` through a semantic conflict with an earlier merge stayed invisible until it surfaced inside somebody else's unrelated PR. All three now also run on push to `main`. The linter ran pint inside the unpinned `composer` image. This codebase uses PHP 8.5 syntax — `new Stream\Factory()->createStream(...)` without parens, typed class constants — which pint can only parse on a new enough runtime, so whichever PHP `composer:latest` happened to ship was an unstated build dependency: the job would break, or quietly under-lint, whenever the image moved. Everything else here pins carefully (`composer:2.7`, `redis:7.2-alpine`, `utopia-base:php-8.5-2.1.0`, checkout by SHA); this was the one exception. It now runs against the project's own test image, the way the analysis job already does, which pins the runtime and drops the duplicate dependency install path with it. Verified by running the job's exact commands. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The consuming side is deliberately interoperable: `Remote::event()` accepts
any non-empty id, http-feeds endpoints commonly use UUIDs, and `consume()`
tracks such a position and sends it back as `lastEventId` without
complaint. But `seek()` gated on `Id::isValid()`, which describes the
`{ms}-{seq}` shape a Redis stream mints — a producer-side implementation
detail.
Against a third-party feed everything therefore worked right up until the
day an operator needed the documented poison-event escape hatch, and then
`seek($poisonEventId)` rejected the very id the consumer had just handled
and saved. There was no workaround short of writing to the cursor store by
hand. The same inconsistency from the other side: `consume()` saved those
ids happily, so "positions must be well formed" was only ever enforced on
the manual path.
The rule is now the feed's to state rather than this library's. A local
`Store` mints its positions and pages by decoding them, so a malformed id
is a caller's mistake and is still refused. Any other `Readable` — a
`Remote`, or someone's own implementation — is the authority on its ids,
so any non-empty one is accepted. Both still refuse the tip sentinel,
which stands for wherever the feed ends when the request arrives: a start,
not a position, and already expressible as `reset()` on a START_TIP
consumer.
The shared consumer scenarios split accordingly: ids no feed could use
stay asserted everywhere, shape judgements run only where the feed owns
the shape, and `Consumer/HttpTest` gains the case this is all about — a
UUID seek that persists.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The docblock promised `Invalid` when `lastEventId` is present but is neither a position nor the tip sentinel, but the coercion ran first and folded every non-string into `null`. PHP parses `?lastEventId[]=1-0` into an array, so that request was *present*, was *not* a position, and read as "from the oldest retained event". The result was the worst possible answer to a malformed parameter: instead of a 400, the caller got a full replay of the retained feed — the most expensive response the endpoint has — and a caught-up consumer would read it as a sudden flood of new events rather than as the error it was. `limit` and `timeout` deliberately keep falling back to their defaults for the same input. Both are the producer's to decide and neither can cause a wrong answer, which is the existing "garbage limit" policy; a comment and a test now say that the difference is a choice rather than an oversight. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`publish()` takes a "prepared `CloudEvent`" and rebuilds it with the producer's own `source`, discarding whatever the caller set. That is deliberate and tested, but nothing said so: the docblock listed only the exceptions, and the README did not mention `publish()` at all. A caller relaying an event received from another feed would reasonably expect a prepared event to be published as prepared, and instead had its origin silently rewritten. The docblock and the README now name all three attributes the producer owns — `source`, `id` (the store assigns it, since it is also the event's position) and a missing `time` — and say what to do about the relay case: keep the original origin in an extension attribute. Writing it down surfaced a fourth. `specversion` is passed through by `publish()` but silently normalised to `1.0` by the store round trip, since that is the only version `CloudEvent::fromArray()` accepts — keeping another one would leave an entry in the feed that nothing could read. Documented and tested rather than left as a claim, along with the source replacement itself. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`application/cloudevents-batch+json` was written out on both sides of the wire: `Batch::MEDIA_TYPE` for the `Content-Type` a feed response carries, `Remote::MEDIA_TYPE` for the `Accept` every read sends. They are one contract — one side's Content-Type is literally the other side's Accept — and `Readable` is where this library already keeps what the serving and consuming sides share. Editing one copy, say for a parameterized variant, would have desynchronized the pair silently; only a test comparing the two literals would have caught it, and there was none. The value now lives on `Readable`. `Remote` implements the interface so it inherits the constant outright, and `Batch::MEDIA_TYPE` stays as an alias of it: both call sites keep the name that reads naturally where they are used, while the drift is structurally impossible rather than merely tested. The README's route example uses `Batch::MEDIA_TYPE` instead of repeating the literal a third time. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The library requires PHP >= 8.5 but tested with PHPUnit 9.6, which went end of life in early 2024 — four majors behind. A brand-new library on a brand-new PHP version has the longest runway to be broken by a point release, and the least chance of a fix arriving for a test framework nobody supports any more. The migration also only ever gets more expensive: `phpunit.xml` used PHPUnit-9-only attributes (`convertErrorsToExceptions`, `backupStaticAttributes`) and the tests carried metadata in doc comments (`@dataProvider`), which PHPUnit 10 dropped. Every test added before the move would have added to it. Doing it at the initial PR is as cheap as it will ever be. All 13 `@dataProvider` annotations became `#[DataProvider]` attributes — the providers were already `static`, so nothing else had to change — and `phpunit.xml` was regenerated via `--migrate-configuration`, then reformatted back to the file's own layout and given the schema reference so editors can validate it. `.phpunit.cache/`, the result cache's new home, is ignored. Test and assertion counts are unchanged across every suite, which is what says the data providers still feed: a provider PHPUnit no longer sees does not error, it just silently runs the test once with no arguments. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`Server::poll()` clamps with `max(0, min($timeout, MAX_TIMEOUT))` and `serve()` coerces the query parameter before forwarding it. Neither had a test. Every non-zero timeout in the suite went through `poll()` directly, and every `serve()` test passed `'0'` or garbage, so a refactor that dropped the third argument entirely would have left the whole suite green while long polling over HTTP silently became a plain read — invisible in what comes back, since a caught-up read answers with the same empty batch either way. The clamp itself is the security-relevant one: `timeout` arrives from an untrusted HTTP client, so a regression means `?timeout=86400000` holds a worker for as long as the caller asks. The one existing `MAX_TIMEOUT` assertion covers the *consumer's* clamp on its own configuration, which is a different clamp against a different caller. `RecordingStore` captures what the server asked for, since that is the only place clamping is observable, and `ServerTest` pins both parameters across the range: forwarded as given, capped at the protocol maximum, floored (negative timeout to 0, limit to 1), and falling back on garbage — through `serve()` and through `poll()`. A position is asserted to arrive untouched, so the clamping is all `serve()` does. The behavioural half runs per adapter in `Server/Base`: `serve()` with a 600ms timeout on an empty feed must actually wait. Only a lower bound, so it measures the code rather than the CI machine. Confirmed all three fail when `serve()` stops forwarding the timeout. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The README promises `Exception\Transport` when "the backend or network failed: Redis errors, HTTP failures". The HTTP half was tested thoroughly; the Redis half — the flagship production adapter — was not tested at all. `grep RedisException tests/` found nothing, and `FailingCursor` only shows how a consumer reacts to a `Transport` it raises itself, so the actual `\RedisException` → `Transport` wrapping in `Store\Redis::append/tip/read` and `Cursor\Redis::load/save/reset` never ran under test. A regression letting the raw exception out would have shipped green and crashed every consumer catching `Utopia\Feed\Exception` as documented. All six operations are now covered, plus the consumer on top of them. Confirmed by widening the catch so the exception escapes: seven failures, all "RedisException ... does not match Transport". The review suggested closing a connected client, which does not work here — phpredis 6.3 reconnects transparently on the next command, so every call succeeds. A client that was never connected raises `\RedisException` from the extension for every command instead, which is both deterministic and a real misconfiguration rather than a state only a test can reach. Two defensive branches the review listed as dead are now reachable too, and the pair documents a policy rather than an accident. A foreign value under the feed's key makes Redis answer every stream command with an error, returned rather than raised: `append()` must not report a position for an event that is not in the feed, so it raises `Transport`, while `read()` and `tip()` treat an unreadable feed as an empty one — a replay at worst, where failing would stall every consumer. That is the same policy the cache store already applies to a foreign value under its key. `Store\Cache::append`'s false branch, the third one listed, was covered when the cache adapters' wrapping landed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`phpunit.xml` had no include list, so `--coverage-text` could not produce a meaningful report even if run: there was no way to measure coverage on this branch, let alone track it. Every gap the review found — the untested server timeout clamp, the dead Redis error branches, the missing `datacontenttype` round trip — was invisible, and uncovered code in future PRs would land with nothing to notice it. `phpunit.xml` now scopes coverage to `src/`, so a report says what the suite exercises of the library rather than of its own fixtures. The test image carries pcov — line coverage only, which is all a report needs, at a fraction of Xdebug's cost — loaded but switched off, so every ordinary run is unaffected and only `composer coverage` turns it on. Two scripts, because the interesting number and the fast one differ: `coverage` runs the service-free suites (~8s), `coverage:all` adds Redis and pool. CI runs the latter in its own job: the Redis and pooled adapters are exactly the ones whose error paths are easiest to leave untested, so a number that excluded them would flatter the suite. Where it lands today: 97.56% of lines across all suites, 84.11% from the service-free ones alone. No threshold — the number is there to be read, and a badly chosen gate is worse than none. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`grep datacontenttype tests/` used to find nothing, which is exactly why the store round trip could drop it silently: the producer suite carefully round-tripped `subject`, `dataschema`, `time`, extensions and even digits-only extension names, and skipped this one. The bug and the test gap were the same finding from two sides, and fixing one without the other invites the regression straight back. The producer-side round trip landed with that fix. This closes the other two places the review named. `testRoundTripsAnEvent` now asserts the attribute `produce()` sets, so the everyday path carries it rather than only the deliberate `publish()` case. And the `Remote` decode path, which was untested despite `datacontenttype` being in its `ATTRIBUTES` list and read by `optional()`: one event on the wire carrying every context attribute this library models plus an extension, each asserted after decoding. `RemoteTest` previously checked `id` and `data` and nothing else, so a decode that quietly dropped `subject`, `time`, `dataschema` or `datacontenttype` would have passed. A second case pins the absence: a feed that sends no `datacontenttype` decodes without one, rather than having a value invented that the producer never claimed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Several tests asserted an upper bound on elapsed time just above what the correct code takes, which measures the CI runner rather than the code. `testPollHonoursATimeoutShorterThanThePollInterval` allowed 300ms for a 100ms timeout and runs in four adapter suites — in `redis` and `pool` each poll tick is a round trip into a container, on a shared GitHub runner. Two hundred milliseconds of headroom is one noisy neighbour away from red, and that failure mode is the worst kind: rare, unreproducible locally, and it teaches everyone to hit re-run, which is how real regressions get waved through. Each bound is now derived from the elapsed time the regression it guards would produce. Overshooting the deadline means sleeping a full 500ms interval, and ignoring a configured interval means falling back to the 500ms default, so 0.45 catches both while leaving 350–400ms of slack over the ~20–100ms the correct code takes. Same for the no-timeout read, where sleeping at all costs a whole interval. The reasoning is written next to each number so the next person changing one knows what it is protecting. Confirmed by reintroducing both bugs: each fails exactly the test that exists for it, at ~0.50s against the 0.45 bound. Lower bounds are untouched — a sleep cannot finish early, so they cannot flake. `testDelegatesLongPollingToTheProducer` keeps its clock check with a note that the request count next to it is the assertion actually doing the work. Not changed: the deliberate 600ms waits multiplied across five adapter suites. That is run time rather than flakiness, and removing it means a clock abstraction through `Store::poll()` — a change to production code to suit the tests, which is a trade worth making deliberately rather than in passing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`Remote::event()` silently dropped any extension the spec cannot carry — a name outside `[a-z0-9]+`, a value that is not bool/int/string — and none of it was tested. One well-formed extension riding along was covered; what happened to a float, an array, or an uppercase name was nobody's decision, just whatever the code did. The two decode paths also disagreed, and neither side of the disagreement was covered, so nothing would have gone red when it mattered. Over HTTP a foreign `"ratio": 1.5` was filtered out and the event delivered. Read from a local store, the same event was merged verbatim into `CloudEvent::fromArray()`, which rejects it — and since a store read decodes every entry in the batch, that entry failed every read past it, permanently, for every consumer, until it fell off the trim horizon. The worst failure shape a feed has, arrived at by accident. Both now go through one `Extensions::filter()`. Filtering, rather than raising, is the version somebody would have chosen: a feed is read by consumers older than its producer by design, and one odd attribute must not cost the event and everything behind it. This is not hypothetical for a local store either — a Redis stream is writable by any tool, and this library's producer cannot be assumed the only writer, which is how the new Redis test stages it. Tests name each shape the filter drops, so dropping stays a choice: float, array, object and null values; uppercase, dashed and underscored names. And each shape it keeps, including a digits-only name. Fixing that last one turned up the same hazard inside the suite: `RemoteTest::raw()` built fixtures with `array_merge()`, which renumbers integer keys, so a digits-only extension was lost in the fixture before the code under test ever saw it. It unions now. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`testRetentionTrimsToExactlyTheCap` and `testAcceptsTheSmallestUsefulRetentionCap` were copy-pasted line for line into `Producer/MemoryTest` and `Producer/CacheTest`, against the suite's own design principle: one abstract scenario suite per component, extended by every adapter. Two copies drift — tighten one and the other silently stays weak — and a future adapter that trims exactly inherits nothing, having to opt in by copying again. Both now live in `Producer/Base` behind a `trimsExactly()` hook. The property is the default, since it is what every adapter does except the two backed by Redis, where `XADD ... MAXLEN ~` trims to a node boundary — which is the whole reason the shared scenario only asserts the loose bound. `Producer/RedisTest` and `Producer/PoolTest` declare `false` and skip. Confirmed by adapter: Memory and Cache run both, Redis and Pool report them skipped rather than passing vacuously. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`FeedServer` read only the query string. The request path — which carries the feed name `Remote` so carefully `rawurlencode()`s — was discarded, so the fixture answered any name with the one feed it held. A consumer pointed at feed `other` was served `edge` and read the right events by accident; the only name check in the suite was client-side, in `Consumer` rejecting a contradicting `feed:` argument. For a suite whose docblock claims it exercises "the whole contract", routing was the missing half. The endpoint now compares the decoded last path segment against the feed it serves and answers 404 otherwise, and three tests cover what that makes checkable: - A consumer pointed at another feed gets `Transport` with status 404 and records no position. Confirmed to fail without the routing. - A name needing encoding round-trips: `a b/c` is encoded into one path segment and decoded back to the name the producer knows. The existing encoding test asserts the URI string, which cannot show that anything decodes it. - The `Accept` the consumer sends is the `Content-Type` the producer answers with, checked against each other rather than each against a literal — the loop left open when the media type was declared once on `Readable`. `Recorder` captures the served content type for it. Splitting the path before decoding is deliberate and commented: a feed called `a/b` travels as `a%2Fb`, and decoding first would split it into a path nobody asked for. So is tolerating a path with no slash at all — a consumer built over a client with no base URI sends the bare name, which is exactly how this suite builds one. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`Consumer/Base::setUp()` always builds the feed as a `MemoryStore`; the adapter subclasses swap only the cursor. So `Consumer/RedisTest` ran a Redis *cursor* against a *memory* feed, and after the test rework nothing anywhere had a `Consumer` read a real Redis stream. The consumer's paging arithmetic — `Id::after`, strictly-after reads, the batch loop — was exercised only against ids this library mints itself, never against the ones `XADD` assigns or the approximate trimming `XRANGE` reads back. Added to `Consumer/RedisTest` rather than by deriving `Consumer/Base`'s store from the trait, which keeps the change local: the shared scenarios still run against the simplest store there is, so a failure in them is still the consuming side's rather than a store's. Four scenarios, three of them lost with the old `E2E/RedisTest`: consume and resume through a persisted cursor without replaying, two named consumers over one stream, and reset replaying what the stream retains. The fourth is new and is the one the gap was really about — paging a backlog in batches of three. `XADD` assigns ids within a millisecond by bumping the sequence, so a batch boundary regularly falls between two ids sharing a timestamp, which is exactly where paging by string comparison goes wrong. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`Store\Cache` was exercised by the producer suite (store-level reads) and the server suite, but never by a `Consumer`: `Consumer/CacheTest` swapped only the cursor, so the feed it consumed was still a `MemoryStore`. The integration the adapter exists for — a service that already carries a cache keeping both the feed and the position there — was the one thing not covered. Added to `Consumer/CacheTest` rather than by deriving `Consumer/Base`'s store from the trait, matching the Redis change and keeping the shared scenarios pointed at the simplest store there is. The drain-and-restart case is the one lost with the old `StoreCacheTest`. The other two are for how this adapter reads differently from the rest: a caught-up poll tick answers from the tip marker instead of loading the feed, so a test has to prove "caught up" still means caught up and not "nothing more, ever" — and that paging a backlog in batches works over an adapter that scans the whole feed for each page. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Nothing checked `subject`, `source`, `time`, `dataschema` or extensions after an HTTP round trip. `Consumer/Base::drain()` records only `type`, `RemoteTest::testReadsAFeedOverHttp` asserted `id` and `data`, and `BatchTest::testToArrayIsTheWireEncoding` asserted `id` and `specversion`. The producer and server suites do round-trip those attributes, but through the *store* — so an encoder or decoder that dropped one on the wire specifically would pass the whole suite, and the wire is the one place a dropped attribute cannot be recovered from. `Consumer/HttpTest` now pushes one fully populated event through the real code in both directions — `Batch` encoding it in the endpoint, `Remote` decoding it on the way back — and asserts every attribute on the far side, including a nested payload and all three extension value types. That restores what the old `RoundTripTest` covered, and more. `BatchTest` gains the encode side in isolation: one event asserted as a whole array rather than attribute by attribute, so an attribute that stopped being encoded fails rather than going unnoticed. Plus the property that makes the wire format readable — an absent optional attribute is omitted, not sent as null, since the spec has no null attribute values and a consumer has to tell "not set" from "set to nothing". Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The reworked `CursorTest` kept only the `Cursor\None` cases. Most of what the old parameterized suite covered is now exercised indirectly — `Consumer/Base` runs against every cursor adapter and asserts on `load()` throughout — but two things are not reachable from there at all. `Consumer::__construct()` validates its own name before the cursor is ever touched, so the adapters' `Invalid` path had no test outside `None`. And nothing checked that each adapter routes `load`, `save` and `reset` through the shared `Cursor::key()` rather than building a key of its own — an adapter that did would skip the validation and drift from the documented `feed:<feed>:cursor:<consumer>` layout at the same time, and only Redis has a layout test. `tests/Feed/Cursor/Base.php` follows the same shape as the other suites, subclassed for Memory, Cache, Redis and Pool, each joining its adapter's existing CI job. It covers the direct API the consumer scenarios reach only sideways: unknown consumer, save/load, overwrite, isolation by consumer and by feed, reset, reset of a position that was never saved, and reset touching only the consumer it names. The key assertion is the last one — all three operations against all three unusable name combinations, nine cases per adapter. Confirmed load-bearing by making `Cursor\Memory` build its own key: all nine fail. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The file had two `## 0.1.0` sections, the top one documenting breaking changes over the bottom one, which was labelled "Initial release". One version cannot both be the initial release and break it. The top section also documented APIs that do not exist. `Start::Tip` is `Consumer::START_TIP`, a string constant rather than an enum. `Protocol::MAX_BATCH`/`MAX_TIMEOUT` live on `Readable`. `Protocol::encode()`, `Protocol::decode()` and `Protocol::MEDIA_TYPE` have no class at all — those responsibilities are `Batch`'s and `Remote`'s — so the note about a route never needing to name `Protocol` named something a route could not have named anyway. And since this is the first release, migrations between versions nobody has ever run are not changelog material. Journal → Store and Protocol → Readable/Batch/Remote are design history, which is what git history is for; the reasoning that survived is already in the README. Collapsed into a single `## 0.1.0 — Initial release` describing what actually ships, using the real names, and written after the rest of this branch landed so it documents the API as released rather than as planned — including the seek relaxation, the media type on `Readable`, the cache store's tip marker and its smaller retention default. Every class and constant named here was checked against `src/`, as was every number: 1000, 30s, 500ms, 1 000 and 100 000 entries, 30 days. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The commits from 58cf7ac to 20f0d7c explained themselves at length: several comments argued the case for a decision over three or four sentences where one would do, and a few restated what the line below already said. Every comment added across that range is now either one line, a short paragraph, or gone. What survives is the part a reader cannot get from the code — why `purge()`'s false is not checked, why the tip marker is written before the feed, why the key builder escapes rather than rejects, why a timing bound is the number it is. The rationale that only mattered while the change was being made lives in the commit messages, which is where it belongs. Net −322 lines, no behaviour change: all 669 tests, PHPStan and Pint stay green. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Introduces
utopia-php/feed, a library for moving events between services withpull-based HTTP event feeds (http-feeds.org)
instead of pushing them to every service that needs them.
Where this comes from
Extracted from two implementations of the same idea that grew up on either side
of one feed — a producer in
appwrite-labs/cloud(#5046)and a consumer in
appwrite-labs/edge(#988).Between them they had two event models, two pull loops, two cursor stores, and
two copies of the HTTP contract joining them. The last one is the duplication
that mattered most: the two halves drifting apart is a wire incompatibility, not
a local bug.
docs/migration.mdmaps every existing class in both repos onto this library.Design
JournalJournal\Httpreads another service's journal, so a remote feed and a local one are the same object to everything above it.Feedsource/time, clamps a consumer-suppliedlimit, long-polls.CursorJournal.ConsumerProtocolThe structural point: the producer stores no per-consumer state, which is
what makes adding a consumer free.
Events are
Utopia\CloudEvents\CloudEvent— this library defines no event typeof its own.
Behaviour deliberately preserved from the originals
tip. This is what makes a staged rollout safe.
event retries and everything behind it waits.
feed:<name>:cursor:<consumer>key format and<ms>-<seq>ids, sopositions already handed out survive the migration.
Test plan
Feedis servedthrough
Protocoland consumed throughJournal\Http— both halves of thecontract, not a fixture written to match one side.
confirms: that
XADDids match the format positions are parsed with, and thata consumer below the
MAXLENtrim horizon still reads.All green on PHP 8.5 in CI.
Blocking before release
utopia-php/cloudeventsis pinned to a branch —dev-feat-cloudevents-g2. That branch needs merging and tagging, and thiscomposer.jsonupdating to the tag, before anything here can be released.Review notes
LICENSE(MIT, 2026Utopia.php); this branch also carries
LICENSE.md(MIT, 2013 Eldad Fux)inherited from the utopia repo the scaffolding was based on. One should go —
left in deliberately rather than picking for you.
php-versionsmatrix in.github/workflows/tests.yml; adding a version is one line there.🤖 Generated with Claude Code