Documentation
¶
Overview ¶
Package staterefs stores a large carried state field once instead of copying it into every step that merely passes it along.
A step's state is a full input snapshot, so a field that is carried but not changed is re-serialized into every row it survives into. A field above a size bar is therefore not written into the successor's state at all: it is omitted, and a refs map records which step physically holds the bytes - the ANCHOR.
A Linker holds the size policy and the resolution cache for ONE shard, and has two operations:
stateJSON, refsJSON, err := linker.Mint(merged, changes, inherited, anchorID, successors, inlineOnly) materialized, err := linker.Resolve(ctx, state, refs, nil, load)
Mint is the write side and performs no I/O: the caller's state is already materialized, so omitting a field is all there is to do. Resolve is the read side and reaches the database only through the caller's Loader, which is handed every anchor it needs at once. Both are safe for concurrent use.
The encoding never escapes: Mint takes materialized state and Resolve returns materialized state, so a caller between the two - transition evaluation, a task carrier, an API response - only ever sees literals.
Index ¶
- Constants
- func CombinedReducerFields(state, changes workflow.State, reducers map[string]workflow.Reducer) map[string]bool
- type Anchor
- type Linker
- func (l *Linker) Flatten(ctx context.Context, stateJSON []byte, refs Refs, load Loader) ([]byte, error)
- func (l *Linker) Mint(merged workflow.State, changes workflow.State, inherited Refs, anchorID int, ...) ([]byte, []byte, error)
- func (l *Linker) Resolve(ctx context.Context, state workflow.State, refs Refs, want map[string]bool, ...) (int, error)
- func (l *Linker) ResolveReduced(ctx context.Context, state workflow.State, refs Refs, ...) error
- type Loader
- type Refs
Constants ¶
const Linear = 1
Linear is the successor count of a step with one successor, for callers that would otherwise pass a bare 1 to Mint.
Variables ¶
This section is empty.
Functions ¶
func CombinedReducerFields ¶
func CombinedReducerFields(state, changes workflow.State, reducers map[string]workflow.Reducer) map[string]bool
CombinedReducerFields names the keys whose merged fan-in value came from a COMBINING (non-replace) reducer folding a delta the anchor step wrote onto a base it also held - reduce(state[k], changes[k]). That value exists in NO step row: the anchor's changes column holds only the delta, its state only the base. So it must never be minted as a ref against that anchor (resolution reads changes first and would splice back the bare delta, silently dropping the accumulated base). The fan-in mint therefore passes these as inlineOnly. A key present only in changes (no base) is left off: there merged[k] == changes[k], so the anchor's changes is a sound anchor and keeping the ref preserves the byte win.
Types ¶
type Anchor ¶
type Anchor struct {
State map[string]json.RawMessage
Changes map[string]json.RawMessage
Refs Refs
}
Anchor is one anchor step's payload, as a Loader returns it. Both columns are searched because either can hold a ref'd field's bytes; Refs is what lets Resolve assert that a ref is one hop rather than silently walking a chain.
type Linker ¶
type Linker struct {
// contains filtered or unexported fields
}
Linker mints and resolves state refs for one shard. It is safe for concurrent use.
func New ¶
New creates a Linker for a shard with the given SQL driver name. The driver only corrects the size estimate at a fan-out (see inflation) and is fixed for a shard's life, which is why it is bound here rather than passed per call.
func (*Linker) Flatten ¶
func (l *Linker) Flatten(ctx context.Context, stateJSON []byte, refs Refs, load Loader) ([]byte, error)
Flatten returns a step's state column with every ref'd field spliced back in - the form used where a state snapshot must stand on its own, detached from the anchors that backed it.
It splices json.RawMessage values rather than decoding the state map, so a large payload is never decoded and re-encoded on the way through.
func (*Linker) Mint ¶
func (l *Linker) Mint(merged workflow.State, changes workflow.State, inherited Refs, anchorID int, successors int, inlineOnly map[string]bool) ([]byte, []byte, error)
Mint decides which of a successor step's state fields are stored by reference rather than by value. It returns the JSON to write into the successor's state column (the ref'd fields omitted) and into its refs column.
merged is the successor's fully MATERIALIZED state (refs resolved at dispatch, then the task's changes overlaid), changes is the accumulated delta that produced it, inherited is the DISPATCHED step's own refs, and anchorID is that step - the only step whose row can newly anchor anything here, because every candidate field's bytes are either in its changes (the task just wrote them) or in its state (carried, and not itself a ref). That single-candidate-anchor fact is what makes the free tier free: once any one field opens the anchor, every other field in that row rides along at zero extra read cost.
successors is how many steps are about to carry this state, and it is the primary policy axis - a ~100x swing in the bar between a linear hop and a wide fan-out. Pass Linear for one.
inlineOnly names fields that must be stored as literals whatever their size, because their bytes are in NO step row: a synthesized fan-out element, or a value only this merge produced. Ref'ing one would dangle. It is a distinct signal from a field appearing in changes (which only means the ref must not be CARRIED, and leaves the field free to anchor here).
Minting needs no database read: merged was resolved at dispatch, so every literal is already in hand and "inlining" is simply declining to omit a field.
func (*Linker) Resolve ¶
func (l *Linker) Resolve(ctx context.Context, state workflow.State, refs Refs, want map[string]bool, load Loader) (int, error)
Resolve materializes ref'd fields into state, in place, and reports how many bytes of field value it spliced in. It is the read side of the whole design, and the reason the anchor - not the field - is what policy prices: the Loader fetches whole payload columns, so the cost is one row per DISTINCT anchor (and, on a cache miss, one round trip for all of them together), while any number of fields living in an already-fetched row are free.
The returned count is NOT the Loader's count and the two are not interchangeable - they answer different questions and disagree in both directions. The Loader counts the RAW COLUMNS it scanned, which is the database's byte throughput: it includes the anchor's untouched other fields, and it is ZERO on a cache hit because nothing was read. This count is the bytes that ended up IN state, which is the caller's residency: it excludes the rest of the anchor row, and it is unchanged by a cache hit because the field is materialized either way. Meter throughput with the former and what a carrier holds with the latter.
want, when non-nil, selects which refs to resolve; the rest are left for the caller to carry forward. That is what lets a fan-in reduce the fields it must while a large CARRIED field crosses the cohort as a ref (see ResolveReduced).
func (*Linker) ResolveReduced ¶
func (l *Linker) ResolveReduced(ctx context.Context, state workflow.State, refs Refs, reducers map[string]workflow.Reducer, load Loader) error
ResolveReduced materializes only the ref'd fields a fan-in's reducers actually need, into state in place.
A reducer that COMBINES (append/add/union/merge/min/max/and/or/concat) needs its accumulated base value, so a ref'd field it touches must be materialized or the fold would apply the delta to an absent base and lose everything accumulated so far. The default REPLACE reducer needs nothing: if a branch wrote the field, its literal wins outright (and the stale ref is dropped by the caller's mint); if no branch wrote it, the field is carried and the ref rides forward untouched.
So a field is materialized iff the graph registers a reducer for it. That is a safe superset - a registered replace reducer merely costs a wasted fetch - and it is what keeps a large CARRIED field (the motivating case: a document fanned out over its pages) from being materialized and re-anchored at every fan-in, which would hand back the win in precisely the fan-out graphs the design exists for.
A merely-carried (non-reduced) ref is left un-materialized here and re-emitted onto the fan-in step by Mint's inherited-carry-forward (which handles a ref whose key is absent from the merged state, exactly the case a non-materialized carry produces). That is why this returns no carried set: it is not the caller's to thread - Mint recovers it from the same inherited refs it is already passed.
type Loader ¶
Loader fetches anchors by step id. It is called at most once per Resolve, with every anchor still needed after the cache is consulted, and must return one entry per id it could read.
Taking the whole set at once is not an optimization: the size policy prices a whole anchor ROW (any number of fields living in one already-fetched row are free), which is only true if fetching k anchors costs one round trip rather than k.
A caller metering payload bytes counts them HERE, where the untouched columns are in hand. Summing the decoded per-field values instead would undercount by every key, brace and separator, and would count zero for a column that failed to decode - wrong for a metric whose job is to track a byte-throughput ceiling.
type Refs ¶
Refs maps a state field name to the step that physically holds its bytes.
func Parse ¶
Parse decodes a refs column. An empty or absent value yields no refs and allocates nothing, which is the overwhelmingly common case.
A MALFORMED value also yields no refs rather than an error, and that is a decision rather than convenience. The column is engine-written, so malformed means a bug-state row - but the alternative is worse: every read of that row (Snapshot, Step, History assembly, the fan-in merge, Fork) would fail permanently, bricking a flow that is otherwise intact. Degrading instead loses the ref'd fields, which is the same visible symptom the row already has, and leaves every other operation on the flow working. The loud checks are kept for the cases where the engine still has something to act on: a ref into a missing step, a violated one-hop, and Fork's missing clone mapping all error rather than degrade.
Source Files