Messages stamped by sets of capabilities - #813
Merged
Merged
Conversation
frankmcsherry
force-pushed
the
frames
branch
3 times, most recently
from
August 18, 2026 23:09
11d961b to
58eabaf
Compare
frankmcsherry
changed the title
Replaces the single timestamp on each message with a Stamp<T>: a multiset of timestamps affixed to the message, of which the message's contents may only result in downstream work at times greater or equal to some element. Like postage, the stamp records the capabilities under which the message travels. Messages are accounted in progress tracking once per stamp element, on both the produce and consume sides, so multiplicities are significant; element order is not, and is maintained sorted so that equal stamps are structurally equal. The progress tracker itself is unchanged, as it already consumes multisets of pointstamp updates. This commit is a representation change only: no public interface constructs a stamp with other than exactly one element, every reachable path produces singletons, and singleton stamps follow the same operations as the prior single timestamp. Stamp stores a single element inline, mirroring Antichain's storage, so the common case allocates nothing; the most message-overhead-bound microbenchmark (pingpong, one-element messages, one worker) measures within ~5%, and progress-only benchmarks (barrier) are unchanged. Stamp transformations come in two shapes with distinct obligations, marked by construction: map_pointwise preserves multiplicities and is required at scope boundaries (enter and leave), whose produced and consumed accounting is inferred independently at either end of the channel from the stamp itself and must agree element-wise; map_into restores minimality and is reserved for operators that account for their own messages. Breaking changes: Message.time becomes Message.stamp; Message::push_at takes a Stamp; CapabilityTrait::time() becomes stamp() -> Stamp<T>; Distributor implementations receive a &Stamp<T> and must reproduce it on every produced sub-message; Event::Messages carries a Stamp<T>, so captured event streams from prior versions are not readable; the bincode wire layout of Message changes accordingly. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh
Allows an output session to be opened with a CapabilitySet, stamping its messages with the set's timestamps: a batch of updates can ship whole, stamped with the antichain that justifies its contents, rather than shredded into one message per capability. An empty set stamps messages with no timestamps at all: such messages make no progress claims, are invisible to progress tracking, and sit outside the frontier-ordered delivery guarantee, including best-effort delivery near dataflow teardown. Non-singleton stamps arise only downstream of an explicit CapabilitySet session; no existing program contains one, and unaware code is unaffected. On the receive side, InputCapability::stamp() exposes the stamp and retain_stamp() mints a capability for each element; interfaces that insist on a single timestamp per message (InputCapability::time(), retain(), and capture's Extract) panic with a message directing callers to the stamp interfaces, and can only be provoked by an opt-in upstream in the same dataflow. Feedback advances each stamp element through its summary, discarding elements that cannot traverse and restoring minimality, which is sound there because operators account for their own consumed and produced messages. Tests cover a multi-capability message traversing channels, retain_stamp, and delayed minting; zero-capability delivery after the sender drops all capabilities; multi-stamp messages crossing a data exchange; and a regression test that a stamp leaving a scope remains accounted at one outer pointstamp per element when projection makes elements comparable. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh
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.
Timely messages currently carry exactly one timestamp, acting as the message's capability. This PR generalizes the message stamp to a multiset of timestamps: the message may result in downstream work at times greater or equal to some element, and is accounted in progress tracking once per stamp element (multiplicities are significant; element order is canonical rather than meaningful). Singleton stamps — every existing program — behave identically and are represented inline (mirroring
Antichain's own storage); multi-element stamps let producers ship data stamped by an antichain (e.g. differential batches stamped by their lower bound, without shredding into per-capability tiles); empty stamps let data flow without progress claims.The progress tracker is unchanged: it already consumes multisets of pointstamp updates, and a message stamped by k timestamps is accounted at each of them, on both the produce and consume sides. The changes live in the channel plumbing (
Message, the produce/consume counters, exchange, enter/leave, feedback, capture) and the capability API (CapabilitySetusable as a session capability;InputCapability::stamp()andretain_stamp()).The PR is two commits, in order of increasing adventure:
Non-singleton stamps arise only downstream of an explicit
CapabilitySetsession; no existing code path produces them. Interfaces that assume one timestamp per message (InputCapability::time(),retain(), capture'sExtract) panic with an actionable message if they meet one, which can only be caused by an opt-in upstream in the same dataflow.One invariant deserves reviewer attention: stamps are transformed pointwise at scope boundaries (enter/leave), because boundary accounting is inferred independently at either end of the channel from the stamp itself, and the counts must agree element-wise even when mapping makes elements comparable or equal (as when leaving a scope projects away a timestamp coordinate). Minimizing maps are reserved for operators (feedback) that account for their own messages.
Stamp::map_pointwisevsStamp::map_intoencode the distinction, and a regression test covers the collapsing-leave case.Costs: ~5% on a worst-case per-message microbenchmark (pingpong, single-element messages, one worker); progress-only benchmarks (barrier) unchanged; real payloads amortize further. Benefits (measured in differential, on SCC over concurrently open update rounds): one batch per frontier advance instead of one per capability, 2.7–4.8x end-to-end.
Breaking changes:
Message.time: TbecomesMessage.stamp: Stamp<T>, andMessage::push_attakes aStamp<T>.CapabilityTrait::time()becomesstamp() -> Stamp<T>.Distributorimplementations receive a&Stamp<T>and must reproduce it on every produced sub-message.Event::Messagescarries aStamp<T>: captured event streams persisted by prior versions are not readable, and the bincode wire layout ofMessagechanges accordingly.🤖 Generated with Claude Code
https://claude.ai/code/session_01BrUdeCb6dsunVdk4acCPmh