Deliver scheduled cluster messages at their deadline instead of the next poll - #6452
Deliver scheduled cluster messages at their deadline instead of the next poll#6452sbking wants to merge 5 commits into
Conversation
🦋 Changeset detectedLatest commit: 536f197 The changes in this PR will be included in the next version bump. This PR includes changesets to release 27 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
The NVARCHAR literals cause SQL Server to convert the indexed VARCHAR `shard_id` column, preventing the per-shard index seek.
|
@coderabbitai review |
✅ Action performedReview finished.
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (6)
🚧 Files skipped from review as they are similar to previous changes (5)
📝 WalkthroughWalkthroughAdds ChangesScheduled delivery discovery
Estimated code review effort: 3 (Moderate) | ~25 minutes Sequence Diagram(s)sequenceDiagram
participant StorageInbox
participant MessageStorage
participant StorageWakeHandle
participant StorageReadLatch
StorageInbox->>MessageStorage: Query nextDeliverAt(acquiredShards)
MessageStorage-->>StorageInbox: Return next scheduled timestamp
StorageInbox->>StorageWakeHandle: Schedule delay
StorageWakeHandle->>StorageReadLatch: Re-open at deadline
StorageReadLatch-->>StorageInbox: Resume storage reads
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/effect/src/unstable/cluster/SqlMessageStorage.ts`:
- Around line 406-445: Update getNextDeliverAt so shard IDs are passed as bound
SQL parameters rather than interpolated through wrapString in the mysql, mssql,
and orElse branches. Preserve each dialect’s existing query shape while
replacing the UNION ALL, VALUES, and IN input construction with parameterized
SQL expressions that safely handle embedded quotes.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 0954d953-e1d6-4d68-96f1-99fe7ec9dd98
📒 Files selected for processing (9)
.changeset/cluster-message-storage-next-deliver-at.mdpackages/effect/src/unstable/cluster/MessageStorage.tspackages/effect/src/unstable/cluster/Sharding.tspackages/effect/src/unstable/cluster/SqlMessageStorage.tspackages/effect/test/cluster/MessageStorage.test.tspackages/effect/test/cluster/Sharding.test.tspackages/effect/test/cluster/TestEntity.tspackages/platform-node/test/cluster/MessageStorageTest.tspackages/platform-node/test/cluster/SqlMessageStorage.test.ts
Shard identifiers now reach the nextDeliverAt queries as bound parameters in all three dialect branches, with SQL Server keeping its VARCHAR cast. The storage read loop discovers the next scheduled deadline before reading due messages, so a deadline crossing between the two reads arms a wake instead of waiting for the next poll interval. Regression tests cover a quote-bearing shard group and a deadline that lands between the reads.
tim-smart
left a comment
tim-smart
left a comment
There was a problem hiding this comment.
I'm also not sure about the idea in general. I'll think about it some more.
| */ | ||
| readonly nextDeliverAt: ( | ||
| shardIds: Iterable<ShardId.ShardId> | ||
| ) => Effect.Effect<Option.Option<number>, PersistenceError> |
There was a problem hiding this comment.
It should return a Option<Duration> as the database clock and runner clock could be out of sync.
There was a problem hiding this comment.
Good call. I changed this to return Option<Duration> so storage owns the clock math and Sharding only receives how long to wait. I kept the existing runner-time semantics to match the polling behavior
| pendingNotifications.forEach((entry) => removableNotifications.add(entry)) | ||
| } | ||
|
|
||
| yield* rediscoverStorageWake |
There was a problem hiding this comment.
This should probably be done in the background to keep it out of the read loop.
There was a problem hiding this comment.
Yep, moved this into the background so the read loop doesn’t wait on the extra query. Discovery and the armed timer use separate handles so refreshing the deadline can’t cancel a timer we already know is valid. I added tests for a blocked lookup and for the deadline passing while the lookup is running.
github-project-automation
Bot
moved this from Discussion Ongoing
to Waiting on Author
in PR Backlog
|
@tim-smart That's fair, let me know what you think! I also started this discord thread the other day if you want to chat about the idea there |
3 participants
Summary
Messages saved with a future
deliverAtare only discovered by the periodic storage poll, so a scheduled message can be delivered up to a full poll interval after its deadline.MessageStorage.nextDeliverAtoperation that returns the earliest future scheduled delivery time for a set of shards, with memory and SQL driver implementationsShardingre-arms a single replaceable timer for that deadline; when it fires, it runs the existing storage read pathnextDeliverAtfails, the error is logged at debug level and delivery falls back to the existing pollnextDeliverAtagree whendeliverAtis zeroPostgreSQL and SQLite answer the query with
MIN(deliver_at)over a partial index on(shard_id, deliver_at)filtered to unprocessed scheduled rows. MySQL and SQL Server have no partial indexes, so they seek per shard (correlated scalar subquery andCROSS APPLY ... TOP 1) over a plain(shard_id, processed, deliver_at, last_read)index, which stays flat as processed rows accumulate. Migration0003_deliver_at_indexcreates the index idempotently on all four dialects. #6441 independently adds a0003migration; whichever PR rebases second will renumber its migration.Testing
nextDeliverAtdegrades to poll-cadence delivery with the read loop still usable;deliverAtat epoch zero across the due boundarypnpm vitest run packages/effect/test/cluster/MessageStorage.test.ts packages/effect/test/cluster/Sharding.test.ts packages/effect/test/cluster/Entity.test.ts(42/42)pnpm vitest run packages/platform-node/test/cluster/SqlMessageStorage.test.ts); the SQL Server leg passes on a local branch combining this change with Support SQL Server in cluster SqlMessageStorage #6441's container fixture, with this migration renumbered to0004therepnpm checkpnpm lintBenchmark protocol, reproduction harnesses, and raw results: https://gist.github.com/sbking/0a56b3f94ca518f4e50308f045e75a45
Summary by CodeRabbit
nextDeliverAtto determine the earliest upcoming scheduled delivery time for pending messages across selected shards.deliverAtdeadline, including messages already stored before startup.nextDeliverAtsupport for both in-memory and SQL storage backends.