state

func (*Store) ActivatableJobs ¶

func (s *Store) ActivatableJobs(jobType int32, fn func(jobKey uint64) error) error

ActivatableJobs calls fn with the key of every open job of the given type, via the jobActivatable index — the worker-polling access pattern.

func (*Store) ActivatableJobsDesc ¶

func (s *Store) ActivatableJobsDesc(jobType int32, before uint64, fn func(jobKey uint64) error) error

ActivatableJobsDesc calls fn with the key of every open job of the given type in DESCENDING key order (newest first), starting just below `before` — before == 0 starts from the newest key. It backs the task inbox's newest-first, cursor-paged listing (GET /tasks?before=): a flood parks its oldest tasks at the low keys, so paging from the newest downward surfaces the tasks an operator most likely wants first while still bounding the scan. Worker polling keeps using ActivatableJobs (oldest-first, FIFO) — this is a read-side ordering only.

func (*Store) ActiveElementInstanceCount ¶

func (s *Store) ActiveElementInstanceCount() (int, error)

ActiveElementInstanceCount returns how many element instances are live.

func (*Store) ActiveElementInstances ¶

func (s *Store) ActiveElementInstances(fn func(key uint64, v *model.ElementInstanceValue) error) error

ActiveElementInstances calls fn with the key and value of every live element instance. Each carries the BPMN element (as a compiled-graph index) it sits on, which the live diagram overlay maps back to a diagram element.

func (*Store) ActiveProcessInstanceCount ¶

func (s *Store) ActiveProcessInstanceCount() (int, error)

ActiveProcessInstanceCount returns how many process instances are live.

func (*Store) ActiveProcessInstances ¶

func (s *Store) ActiveProcessInstances(fn func(key uint64, v *model.ProcessInstanceValue) error) error

ActiveProcessInstances calls fn with the key and value of every live process instance, via the process-instance column family — the operator "list running instances" access pattern.

func (*Store) AllActivatableJobs ¶

func (s *Store) AllActivatableJobs(fn func(jobKey uint64) error) error

AllActivatableJobs calls fn with the key of every open job of ANY type, via the same jobActivatable index — the whole column family rather than one job type's slice. It backs the read side of the operator complete/fail affordance (listing the jobs an instance is parked on); worker polling still uses the type-scoped ActivatableJobs.

func (*Store) Close ¶

func (s *Store) Close() error

Close flushes and closes the store.

func (*Store) CompensablesOfScope ¶

func (s *Store) CompensablesOfScope(scope uint64, fn func(v *model.CompensableValue) error) error

CompensablesOfScope calls fn with every completed compensable activity retained under the given scope, in completion order (ADR-0103). Used to surface a scope's pending compensations to operators and by tests to assert the index is cleaned when a scope tears down. Mirrors VariablesOfScope (committed reads only).

func (*Store) CompletedProcessInstances ¶

func (s *Store) CompletedProcessInstances(fn func(key uint64, v *model.ProcessInstanceValue) error) error

CompletedProcessInstances calls fn with the key and value of every process instance that has reached a terminal state, via the history column family — the operator "list finished instances" access pattern (ADR-0017). Each value carries its terminal State and CompletedAt.

func (*Store) DataObjectSnapshotHistory ¶

func (s *Store) DataObjectSnapshotHistory(scopeKey uint64, fn func(ts int64, pos uint64, v *model.DataObjectValue) error) error

DataObjectSnapshotHistory folds the retained data-object state changes of one scope, calling fn with each change's event timestamp, log position, and the object's new state in the order they occurred (ADR-0053). Because the key sorts by timestamp then position, a scope-wide scan yields a monotonic sequence a caller folds by position — the event-sourced data-state timeline and the basis for lineage. Mirrors VariableSnapshotHistory (ADR-0048).

func (*Store) DataObjectsOfScope ¶

func (s *Store) DataObjectsOfScope(scope uint64, fn func(v *model.DataObjectValue) error) error

DataObjectsOfScope calls fn with every data object owned by the given scope, via the data-object column family — the current value of each. Used to surface an instance's data to operators and, later, to build a FEEL scope for data associations (ADR-0053). Mirrors VariablesOfScope.

func (*Store) DecisionEvaluationHistory ¶

func (s *Store) DecisionEvaluationHistory(scopeKey uint64, fn func(ts int64, pos uint64, v *model.DecisionEvaluationValue) error) error

DecisionEvaluationHistory folds the retained DMN decision evaluations of one scope (a process instance), calling fn with each evaluation's event timestamp, log position, and its frozen record (decision id, inputs, outputs, trace) in the order they occurred (ADR-0066). Because the key sorts by timestamp then position, a scope-wide scan yields a monotonic sequence — the same ordering as the variable and element-step timelines, so a business rule task's decision reasoning lines up with the step at which it ran. Used to surface how a decision was made to operators, both live and after the instance has finished.

func (*Store) DefCompletedCount ¶

func (s *Store) DefCompletedCount(procDefKey uint64) (int, error)

DefCompletedCount returns how many instances of one definition have finished (completed or terminated), from the maintained counter in O(1) rather than scanning the history (ADR-0083).

func (*Store) DefInstanceCount ¶

func (s *Store) DefInstanceCount(procDefKey uint64) (int, error)

DefInstanceCount returns how many instances of one definition are live, read from the maintained per-definition counter in O(1) rather than scanning every instance (ADR-0080).

func (*Store) DefLastActivity ¶

func (s *Store) DefLastActivity(procDefKey uint64) (int64, error)

DefLastActivity returns the unix-nano timestamp of a definition's most recent instance lifecycle event (0 if it has had none), read in O(1) (ADR-0083).

func (*Store) DueTimers ¶

func (s *Store) DueTimers(now int64, fn func(timerKey uint64, v *model.TimerValue) error) error

DueTimers calls fn for every timer whose due date is at or before now, in due order. Because the due date is the index prefix, this is a range scan from the start of the timer family up to now — no scheduler structure, no full scan.

func (*Store) EachDecisionEvaluation ¶

func (s *Store) EachDecisionEvaluation(fn func(scopeKey uint64, ts int64, v *model.DecisionEvaluationValue) error) error

EachDecisionEvaluation folds every retained DMN decision evaluation across all process instances, calling fn with the owning scope (process instance) key, the evaluation's event timestamp, and its frozen record (ADR-0066). It scans the whole decision-evaluation column family — the global "which decisions ran, and how often" access pattern the Operations decisions overview uses, as opposed to DecisionEvaluationHistory, which folds a single instance's evaluations. The scan order is scope then timestamp then position; a caller that only aggregates by decision id does not depend on it.

func (*Store) ElementInstancesOfProcess ¶

func (s *Store) ElementInstancesOfProcess(procKey uint64, fn func(elKey uint64) error) error

ElementInstancesOfProcess calls fn with the key of every element instance belonging to the given process instance, via the elByProc index.

func (*Store) ElementLiveTokens ¶

func (s *Store) ElementLiveTokens(procDefKey uint64, fn func(elementId int32, count int64) error) error

ElementLiveTokens calls fn with each of a definition's elements that currently holds live tokens and how many — one prefix scan over the per-element token counters, so it is O(elements), not O(instances) (ADR-0080).

func (*Store) ElementReplayHistory ¶

func (s *Store) ElementReplayHistory(piKey uint64, fn func(ts int64, pos uint64, v ElementReplayValue) error) error

ElementReplayHistory scans causal token lifecycle facts in deterministic order.

func (*Store) ElementStepHistory ¶

func (s *Store) ElementStepHistory(piKey uint64, fn func(ts int64, pos uint64, elementId int32) error) error

ElementStepHistory folds the retained element-activation steps of one process instance, calling fn with each step's event timestamp, log position, and the activated element's compiled-graph index in the order they occurred (the step-by-step replay timeline, ADR-0046). Because the key sorts by timestamp then position, an instance-wide scan yields a monotonic sequence. The caller resolves the element index to a diagram id via the instance's compiled process.

func (*Store) ElementVisitHistory ¶

func (s *Store) ElementVisitHistory(procDefKey, instanceFilter uint64, fn func(elementId int32, count int64) error) error

ElementVisitHistory folds the token-visit counters for a process definition, calling fn with each visited element index and how many tokens have passed through it. With instanceFilter == 0 it aggregates every instance of the definition (the heatmap the live overlay draws in gray); with a non-zero instanceFilter it reports only that one instance's visits. Because the key ends with the element index and the same element sits under a distinct instance-key prefix per instance, a definition-wide scan can report the same element index once per instance — the caller sums the counts. Pebble folds the merge deltas for each key, so raw carries the current total for that key.

func (*Store) ElementVisitTotals ¶

func (s *Store) ElementVisitTotals(procDefKey uint64, fn func(elementId int32, count int64) error) error

ElementVisitTotals calls fn with each of a definition's elements and its cumulative visit count — the aggregate heatmap, read in O(elements) from the maintained per-element visit counter instead of summing every instance's visit history (ADR-0080, aggregating ADR-0022).

func (*Store) GetElementInstance ¶

func (s *Store) GetElementInstance(key uint64) (*model.ElementInstanceValue, bool, error)

GetElementInstance returns the committed element instance for key, reporting whether it was present. Like GetJob it reads outside a transaction, for consumers such as the in-process DMN worker resolving the decision an activatable business-rule job belongs to.

func (*Store) GetIncident ¶

func (s *Store) GetIncident(elKey uint64) (*model.IncidentValue, error)

GetIncident returns the committed incident on an element instance, or nil if there is none. It reads outside a transaction, for the resolve endpoint's existence check (ADR-0061).

func (*Store) GetJob ¶

func (s *Store) GetJob(key uint64) (*model.JobValue, bool, error)

GetJob returns the committed job for key, reporting whether it was present. Unlike Tx.GetJob it reads outside a transaction, for queries such as a worker runner pulling activatable jobs.

func (*Store) Incidents ¶

func (s *Store) Incidents(fn func(elementKey uint64, v *model.IncidentValue) error) error

Incidents calls fn with the element-instance key and value of every unresolved incident — the operator "list incidents" access pattern (ADR-0061).

func (*Store) InjectCorruptElementInstance ¶

func (s *Store) InjectCorruptElementInstance(key uint64) error

InjectCorruptElementInstance writes an undecodable record under an element instance's key. Like InjectCorruptProcessInstance it is a test/tooling affordance only — it lets a caller in another package exercise the decode-error path of a GetElementInstance read (e.g. the set-variables handler's scope validation). Production code writes element instances through Tx.PutElementInstance, never this.

func (*Store) InjectCorruptIncident ¶

func (s *Store) InjectCorruptIncident(elKey uint64) error

InjectCorruptIncident writes an undecodable record under an incident's key — the incident-list counterpart of InjectCorruptProcessInstance. It lets a caller exercise the decode-error branch of the incident scan (Incidents), which the operator "what's stuck" list depends on to surface a 500 rather than silently drop rows. Test/tooling only; production writes incidents through Tx.PutIncident.

func (*Store) InjectCorruptProcessInstance ¶

func (s *Store) InjectCorruptProcessInstance(key uint64) error

InjectCorruptProcessInstance writes an undecodable record under a process instance's key. It is a test/tooling affordance only — it lets a caller in another package exercise the decode-error path of the active-instance scan (ActiveProcessInstances) that operator read/admin handlers depend on. Production code writes process instances through Tx.PutProcessInstance, never this.

func (*Store) JobOfElement ¶

func (s *Store) JobOfElement(elKey uint64) (uint64, bool, error)

JobOfElement returns the key of the job held by the given element instance, or ok=false if it holds none. It is the read-side counterpart of Tx.JobOfElement, used to resolve one instance's open jobs through the element→job reverse index rather than scanning the global activatable index.

func (*Store) LastAppliedPosition ¶

func (s *Store) LastAppliedPosition() (uint64, error)

LastAppliedPosition returns the highest log position folded into committed state, or 0 if none has been applied yet (genesis).

func (*Store) MessageFlowHistory ¶

func (s *Store) MessageFlowHistory(receiverDefKey uint64, fn func(ts int64, pos uint64, v *model.MessageFlowValue) error) error

MessageFlowHistory folds the retained message flows a definition received, calling fn with each flow's event timestamp, log position, and payload in the order they occurred (the replay timeline). Because the key sorts by timestamp then position, a definition-wide scan yields a monotonic sequence. The caller resolves the receiver element index to a diagram id via the compiled process.

func (*Store) NewTransaction ¶

func (s *Store) NewTransaction() *Tx

NewTransaction starts a transaction. Reads through it see its own pending writes (it is an indexed batch). Mutations become visible only on Commit. The underlying batch is drawn from the store's cache when available, so steady- state processing does not allocate one (invariant I1).

func (*Store) ProcessInstance ¶

func (s *Store) ProcessInstance(key uint64) (*model.ProcessInstanceValue, bool, error)

ProcessInstance returns the process instance for key and whether it was found, looking first in the active family and then in the terminal-history family (ADR-0017). It lets a query resolve an instance's definition whether it is still running or already finished — the lookup the single-process replay uses.

func (*Store) StartTimers ¶

func (s *Store) StartTimers(fn func(timerKey uint64, v *model.TimerValue) error) error

StartTimers calls fn for every armed start timer — one whose owning process instance key is zero, so it instantiates a definition on fire rather than continuing a waiting element (ADR-0051). It is a full scan of the timer family, used only when a definition is (re)deployed (off the hot path), so arming can be idempotent and can supersede a prior version's schedule.

func (*Store) VariableAuditHistory ¶

func (s *Store) VariableAuditHistory(piKey uint64, fn func(ts int64, pos uint64, v *model.VariableAuditValue) error) error

VariableAuditHistory folds the retained external variable overrides of one process instance, calling fn with each override's event timestamp, log position, and its frozen record (who set which variable, on which scope, to what value) in the order they occurred (ADR-0098). Because the key sorts by timestamp then position, a scope-wide scan yields a monotonic sequence — the same ordering as the variable and decision timelines — so the "who changed it" trail lines up with the step at which each override happened. It surfaces to operators both live and after the instance has finished, since the records are append-only history.

func (*Store) VariableSnapshotHistory ¶

func (s *Store) VariableSnapshotHistory(scopeKey uint64, fn func(ts int64, pos uint64, v *model.VariableValue) error) error

VariableSnapshotHistory folds the retained variable changes of one scope (a process instance), calling fn with each change's event timestamp, log position, and the variable's new state in the order they occurred (ADR-0048). Because the key sorts by timestamp then position, a scope-wide scan yields a monotonic sequence a caller folds by position to reconstruct the variables as of any step.

func (*Store) VariablesOfScope ¶

func (s *Store) VariablesOfScope(scope uint64, fn func(v *model.VariableValue) error) error

VariablesOfScope calls fn with every variable owned by the given scope, via the variable column family. Used to build a FEEL evaluation scope and to surface an instance's variables to operators.

func (*Tx) ActiveChildren ¶

func (t *Tx) ActiveChildren(scope uint64) (int32, error)

ActiveChildren returns the active-child count for scope (0 if none). This read folds the merged deltas, so it is used only where the current count is needed (e.g. detecting a finished scope), not on every increment.

func (*Tx) ActiveStartKeyCount ¶

func (t *Tx) ActiveStartKeyCount(defKey uint64, correlationKey string) (int32, error)

ActiveStartKeyCount returns how many live instances of defKey began with correlationKey (0 if none). It folds the merged deltas, so it is read only where the current count is needed — the singleton-start gate (ADR-0094), not on every merge.

func (*Tx) Close ¶

func (t *Tx) Close() error

Close releases the transaction, returning its batch to the store for reuse. Safe to call after Commit. The Tx must not be used afterward.

func (*Tx) Commit ¶

func (t *Tx) Commit() error

Commit applies the transaction. It does not fsync: durability is the WAL's responsibility (ADR-0005), and the store is rebuildable by replay, so a state commit lost to a crash is simply re-derived on recovery.

func (*Tx) CompensablesOfScopeDesc ¶

func (t *Tx) CompensablesOfScopeDesc(scopeKey uint64, fn func(seq uint64, v *model.CompensableValue) error) error

CompensablesOfScopeDesc calls fn for every completed compensable activity recorded under scopeKey, newest first (reverse completion order) — the order a compensation throw runs handlers in (ADR-0103). It reads through the in-flight batch, so it observes records written earlier in the same batch. seq is the record's key sequence (log position), which the caller carries on the consume event to delete it.

func (*Tx) CorrelatableSubscriptions ¶

func (t *Tx) CorrelatableSubscriptions(name, correlationKey string, fn func(elKey uint64, v *model.MessageSubscriptionValue) error) error

CorrelatableSubscriptions calls fn for every open subscription matching the given (message name, correlation key), via a prefix scan — the publish access pattern. It reads through the in-flight batch, so it observes subscriptions created earlier in the same batch (ADR-0020).

func (*Tx) DecDefInstanceCount ¶

func (t *Tx) DecDefInstanceCount(procDefKey uint64) error

func (*Tx) DecElementToken ¶

func (t *Tx) DecElementToken(procDefKey uint64, elementId int32) error

func (*Tx) DecrementActiveChildren ¶

func (t *Tx) DecrementActiveChildren(scope uint64) error

DecrementActiveChildren removes one active child from scope. A scope that returns to zero leaves a zero-valued counter entry rather than being deleted; completion checks treat absent and zero alike.

func (*Tx) DecrementActiveStartKey ¶

func (t *Tx) DecrementActiveStartKey(defKey uint64, correlationKey string) error

func (*Tx) DeleteCanceling ¶

func (t *Tx) DeleteCanceling(txKey uint64) error

DeleteCanceling drops the cancelling marker for txKey when the transaction tears down (ADR-0108). Idempotent — a transaction that was never cancelled, or a plain subprocess that never carried a marker, is a no-op.

func (*Tx) DeleteCompensable ¶

func (t *Tx) DeleteCompensable(scopeKey, seq uint64) error

DeleteCompensable removes one compensable record (its activity has been compensated), located by its scope and sequence — both carried on the consume event, so recovery deletes the identical entry. Idempotent.

func (*Tx) DeleteCompensablesOfScope ¶

func (t *Tx) DeleteCompensablesOfScope(scopeKey uint64) error

DeleteCompensablesOfScope drops every compensable record held under a scope, called when the scope tears down (its subprocess container or process instance completes or is terminated) so uncompensated records never leak past the scope (ADR-0103). Keys are collected before deleting so the scan is not disturbed. Idempotent — a scope with none is a no-op.

func (*Tx) DeleteElementInstance ¶

func (t *Tx) DeleteElementInstance(key uint64, v *model.ElementInstanceValue) error

DeleteElementInstance removes the element instance and its index entry. The value is required to locate the elByProc entry; on recovery it comes from the event payload.

func (*Tx) DeleteIncident ¶

func (t *Tx) DeleteIncident(elKey uint64) error

DeleteIncident removes the incident attached to an element instance. Deleting one that is absent is a harmless no-op — how terminating an element clears any incident it carried without first reading it (ADR-0061).

func (*Tx) DeleteJob ¶

func (t *Tx) DeleteJob(key uint64, v *model.JobValue) error

DeleteJob removes the job, its activatable index entry, and the reverse element→job entry. Errors are accumulated across the three batch deletes (first non-nil wins), keeping every delete on the same covered path.

func (*Tx) DeleteMessageSubscription ¶

func (t *Tx) DeleteMessageSubscription(v *model.MessageSubscriptionValue) error

DeleteMessageSubscription removes a subscription. The value supplies the name, correlation key, and element-instance key that locate its index entry; on recovery they come from the event payload.

func (*Tx) DeleteProcessInstance ¶

func (t *Tx) DeleteProcessInstance(key uint64) error

DeleteProcessInstance removes the process instance.

func (*Tx) DeleteSignalSubscription ¶

func (t *Tx) DeleteSignalSubscription(v *model.SignalSubscriptionValue) error

DeleteSignalSubscription removes a signal subscription. The value supplies the name and element-instance key that locate its index entry; on recovery they come from the event payload.

func (*Tx) DeleteTimer ¶

func (t *Tx) DeleteTimer(key uint64, v *model.TimerValue) error

DeleteTimer removes the timer. The value supplies the due date that locates its index key; on recovery it comes from the event payload.

func (*Tx) DeleteVariable ¶

func (t *Tx) DeleteVariable(scope uint64, name string) error

DeleteVariable removes a variable from its scope by name. It is idempotent — deleting an absent variable is a no-op — and is used to drop an activity-local variable scope when the activity completes (ADR-0068).

func (*Tx) ElementInstancesOfProcess ¶

func (t *Tx) ElementInstancesOfProcess(procKey uint64, fn func(elKey uint64, v *model.ElementInstanceValue) error) error

ElementInstancesOfProcess calls fn for every element instance of a process instance visible in this transaction — committed rows plus this batch's own writes — via the elByProc index. A parallel join uses it to count how many tokens have arrived on its incoming flows, including one activated earlier in the same batch (which a committed-only store scan would miss).

func (*Tx) GetDataObject ¶

func (t *Tx) GetDataObject(scope uint64, name string) (*model.DataObjectValue, error)

GetDataObject returns a scope's data object by name, or nil if absent.

func (*Tx) GetElementInstance ¶

func (t *Tx) GetElementInstance(key uint64) (*model.ElementInstanceValue, error)

GetElementInstance returns the element instance, or nil if absent.

func (*Tx) GetElementInstanceInto ¶

func (t *Tx) GetElementInstanceInto(key uint64, dst *model.ElementInstanceValue) (bool, error)

GetElementInstanceInto decodes the element instance into dst without allocating, reporting whether it was present.

func (*Tx) GetIncident ¶

func (t *Tx) GetIncident(elKey uint64) (*model.IncidentValue, error)

GetIncident returns the incident attached to an element instance, or nil.

func (*Tx) GetJob ¶

func (t *Tx) GetJob(key uint64) (*model.JobValue, error)

GetJob returns the job, or nil if absent.

func (*Tx) GetJobInto ¶

func (t *Tx) GetJobInto(key uint64, dst *model.JobValue) (bool, error)

GetJobInto decodes the job into dst without allocating, reporting whether it was present.

func (*Tx) GetProcessInstance ¶

func (t *Tx) GetProcessInstance(key uint64) (*model.ProcessInstanceValue, error)

GetProcessInstance returns the process instance, or nil if absent.

func (*Tx) GetProcessInstanceInto ¶

func (t *Tx) GetProcessInstanceInto(key uint64, dst *model.ProcessInstanceValue) (bool, error)

GetProcessInstanceInto decodes the process instance into dst without allocating, reporting whether it was present.

func (*Tx) GetVariable ¶

func (t *Tx) GetVariable(scope uint64, name string) (*model.VariableValue, error)

GetVariable returns a scope's variable by name, or nil if absent.

func (*Tx) InboundHighWater ¶

func (t *Tx) InboundHighWater(sourceID string) (uint64, error)

InboundHighWater returns the last-applied sequence for a source, or 0 if the source has no mark yet. It reads through the in-flight batch, so a guard sees a mark written earlier in the same batch.

func (*Tx) IncDefCompletedCount ¶

func (t *Tx) IncDefCompletedCount(procDefKey uint64) error

IncDefCompletedCount bumps a definition's finished-instance count by one, on each process-instance completion or termination. Monotonic (never decremented) — the count of finished instances only grows — so the summary's "finished" column reads in O(1) instead of scanning the history, which draining active instances only makes larger (ADR-0083).

func (*Tx) IncDefInstanceCount ¶

func (t *Tx) IncDefInstanceCount(procDefKey uint64) error

IncDefInstanceCount and DecDefInstanceCount move a definition's active-instance count by one, on process-instance creation and termination.

func (*Tx) IncElementToken ¶

func (t *Tx) IncElementToken(procDefKey uint64, elementId int32) error

IncElementToken and DecElementToken move a definition-element live-token count by one, on element-instance activation and completion/termination.

func (*Tx) IncElementVisitAgg ¶

func (t *Tx) IncElementVisitAgg(procDefKey uint64, elementId int32) error

IncElementVisitAgg bumps a definition-element cumulative-visit count on activation. Never decremented — it is the retained historical heatmap.

func (*Tx) IncrementActiveChildren ¶

func (t *Tx) IncrementActiveChildren(scope uint64) error

IncrementActiveChildren adds one active child to scope. It is a write-only merge (no read), so it does not allocate on the hot path (invariant I1).

func (*Tx) IncrementActiveStartKey ¶

func (t *Tx) IncrementActiveStartKey(defKey uint64, correlationKey string) error

IncrementActiveStartKey / DecrementActiveStartKey maintain the count of live message-start instances of a definition that began with a correlation key (ADR-0094). Like the active-children counter they are write-only composing merges, so they neither read nor allocate beyond the reused scratch buffer, and rebuild identically on replay (I4/I6).

func (*Tx) IsCanceling ¶

func (t *Tx) IsCanceling(txKey uint64) (bool, error)

IsCanceling reports whether the transaction scope txKey was marked cancelling by a cancel end event (ADR-0108). It reads through the in-flight batch, so it observes a mark written earlier in the same batch.

func (*Tx) JobOfElement ¶

func (t *Tx) JobOfElement(elKey uint64) (uint64, bool, error)

JobOfElement returns the key of the job held by the given element instance, or ok=false if it holds none. Used to cancel a host activity's job when an interrupting boundary event terminates it.

func (*Tx) PutDataObject ¶

func (t *Tx) PutDataObject(v *model.DataObjectValue) error

PutDataObject writes (upserts) a data object under its scope and name — the current value, mirroring PutVariable. The live store keeps only the latest; the whole state history lives in the snapshot family (ADR-0053).

func (*Tx) PutElementInstance ¶

func (t *Tx) PutElementInstance(key uint64, v *model.ElementInstanceValue) error

PutElementInstance writes the element instance and its elByProc index entry.

func (*Tx) PutInboundHighWater ¶

func (t *Tx) PutInboundHighWater(sourceID string, seq uint64) error

PutInboundHighWater upserts an external event source's last-applied sequence. It writes the absolute sequence (not a delta), so a replayed IntentInbound- DeliveryApplied rebuilds the identical mark (invariant I4). Reads through the batch see it immediately, so the guard in the same batch observes it.

func (*Tx) PutIncident ¶

func (t *Tx) PutIncident(v *model.IncidentValue) error

PutIncident writes an incident, keyed by the element instance it is attached to.

func (*Tx) PutJob ¶

func (t *Tx) PutJob(key uint64, v *model.JobValue) error

PutJob writes the job, its activatable index entry, and the reverse element→job entry (so an interrupting boundary event can find the host's job). The writes go to one in-memory batch; their errors are accumulated (first non-nil wins) rather than checked one at a time, which keeps every write on the same covered path.

A job is on the activatable index iff it has retries left (Retries > 0): a job whose retries are exhausted stays stored — an incident points at it — but is never handed to a worker until an operator resolves the incident and restores a positive retry count (ADR-0061).

func (*Tx) PutMessageSubscription ¶

func (t *Tx) PutMessageSubscription(v *model.MessageSubscriptionValue) error

PutMessageSubscription writes an open message subscription, keyed by its (name, correlationKey) match pair plus its element-instance key.

func (*Tx) PutProcessInstance ¶

func (t *Tx) PutProcessInstance(key uint64, v *model.ProcessInstanceValue) error

PutProcessInstance writes the process instance.

func (*Tx) PutProcessInstanceHistory ¶

func (t *Tx) PutProcessInstanceHistory(key uint64, v *model.ProcessInstanceValue) error

PutProcessInstanceHistory records a terminal (completed/terminated) process instance in the history index. Written from applyToState when an instance ends, from the event alone, so it replays identically on recovery (ADR-0017).

func (*Tx) PutSignalSubscription ¶

func (t *Tx) PutSignalSubscription(v *model.SignalSubscriptionValue) error

PutSignalSubscription writes an open signal subscription, keyed by its signal name plus its element-instance key. A signal has no correlation key — it matches by name alone (ADR-0088).

func (*Tx) PutTimer ¶

func (t *Tx) PutTimer(key uint64, v *model.TimerValue) error

PutTimer writes the timer into the due-date index, which is its primary store.

func (*Tx) PutVariable ¶

func (t *Tx) PutVariable(v *model.VariableValue) error

PutVariable writes (upserts) a process variable under its scope and name.

func (*Tx) RecordCompensable ¶

func (t *Tx) RecordCompensable(pos uint64, v *model.CompensableValue) error

RecordCompensable retains one completed compensable activity under its scope, keyed by the completion event's log position so a scope scan yields completion order. pos comes from the event header; the value carries the scope, the compensated activity, and its compensation handler.

func (*Tx) RecordDataObjectSnapshot ¶

func (t *Tx) RecordDataObjectSnapshot(ts int64, pos uint64, v *model.DataObjectValue) error

RecordDataObjectSnapshot retains one data-object state change under its scope, keyed in change order. ts and pos come from the event header; the value is the object's new state (name, data state, value). Written only from applyToState, from the event alone, so it rebuilds identically on replay (invariant I4); a plain Set on a unique (position-bearing) key, never overwritten (ADR-0053).

func (*Tx) RecordDecisionEvaluation ¶

func (t *Tx) RecordDecisionEvaluation(ts int64, pos uint64, v *model.DecisionEvaluationValue) error

RecordDecisionEvaluation retains one decision evaluation under its owning process instance, keyed in evaluation order. ts and pos come from the event header; the value carries the decision id, input context, outputs, and trace.

func (*Tx) RecordElementReplay ¶

func (t *Tx) RecordElementReplay(piKey uint64, ts int64, pos uint64, elementID int32, elementKey, tokenID, parentTokenID uint64, sourceFlowID int32, action byte) error

RecordElementReplay retains an activation or consumption with its durable token lineage. It is derived only from the lifecycle event by applyToState.

func (*Tx) RecordElementStep ¶

func (t *Tx) RecordElementStep(piKey uint64, ts int64, pos uint64, elementId int32) error

RecordElementStep retains one element activation of a process instance under its instance key, keyed in time order. ts and pos come from the event header; the value is the activated element's compiled-graph index.

func (*Tx) RecordElementVisit ¶

func (t *Tx) RecordElementVisit(procDefKey, piKey uint64, elementId int32) error

RecordElementVisit adds one to the visit counter for an element instance's element. Called from applyToState when an element instance is activated.

func (*Tx) RecordMessageFlow ¶

func (t *Tx) RecordMessageFlow(ts int64, pos uint64, v *model.MessageFlowValue) error

RecordMessageFlow retains one delivered message flow under its receiver definition, keyed in time order. ts and pos come from the event header.

func (*Tx) RecordVariableAudit ¶

func (t *Tx) RecordVariableAudit(ts int64, pos uint64, v *model.VariableAuditValue) error

RecordVariableAudit retains one external variable override under its owning process instance, keyed in change order (ADR-0098). ts and pos come from the event header; the value carries who set the variable, on which scope, and to what value. Written only from applyToState, from the event alone, so it rebuilds identically on replay (invariant I4); a plain Set on a unique (position-bearing) key, never overwritten.

func (*Tx) RecordVariableSnapshot ¶

func (t *Tx) RecordVariableSnapshot(ts int64, pos uint64, v *model.VariableValue) error

RecordVariableSnapshot retains one variable change under its scope, keyed in change order. ts and pos come from the event header; the value is the variable's new state (name, kind, value).

func (*Tx) SetCanceling ¶

func (t *Tx) SetCanceling(txKey uint64) error

SetCanceling marks the transaction scope txKey as cancelling: a cancel end event fired in it, so when its scope drains the transaction routes out its cancel boundary rather than completing normally (ADR-0108). It is derived in applyToState from the cancel end event's committed Completed event, so it rebuilds identically on replay (I4/I6). The value is a single non-empty byte; only presence is meaningful.

func (*Tx) SetDefLastActivity ¶

func (t *Tx) SetDefLastActivity(procDefKey uint64, unixNano int64) error

SetDefLastActivity records a definition's most recent instance-event timestamp by overwrite (ADR-0083). The processor's event timestamps are non-decreasing in log order, so the last write is the latest and replay rebuilds the identical value (invariant I4). Write-only, no read.

func (*Tx) SetLastAppliedPosition ¶

func (t *Tx) SetLastAppliedPosition(pos uint64) error

SetLastAppliedPosition records, within this transaction, the highest log position folded into state. Committed atomically with the mutations so state and position never diverge.

func (*Tx) SubscribedSignals ¶

func (t *Tx) SubscribedSignals(name string, fn func(elKey uint64, v *model.SignalSubscriptionValue) error) error

SubscribedSignals calls fn for every open subscription waiting on the given signal name, via a name-only prefix scan — the broadcast access pattern. Like CorrelatableSubscriptions it reads through the in-flight batch, so it observes subscriptions created earlier in the same batch (ADR-0088).

func (*Tx) VariablesOfScope ¶

func (t *Tx) VariablesOfScope(scope uint64, fn func(v *model.VariableValue) error) error

VariablesOfScope calls fn with every variable owned by scope, via a prefix scan over the variable column family. It reads through the in-flight batch, so it observes variables written earlier in the same batch. A message throw event uses it to gather the payload it publishes (ADR-0035).