engine

package
v0.6.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 9, 2026 License: AGPL-3.0 Imports: 14 Imported by: 0

Documentation

Overview

Package engine is the heart of Atlas: a single-writer processor that folds commands into durable events and applies them to state.

One partition is driven by one goroutine (invariant I3), so there are no locks on process state. Each batch follows the fixed order append → one fsync → commit state → side effects (invariants I2, ADR-0005). State changes from a record happen in exactly one place, applyToState, used identically live and on recovery (invariant I4), which is what makes crash recovery a simple replay.

The processor path is allocation-free per command and per event (invariant I1): payloads flow by value (see inflightValue), and the batch buffers, queue, side-effect list, and encode buffer are reused across batches. State reads (which decode from the store) and the per-batch state transaction are the remaining allocation sources, tracked separately.

Index

Constants

View Source 
const AgentCallIdVariable = "toolCallId"

AgentCallIdVariable is the variable an activated tool carries in its own scope naming the call it answers, so a result can be paired back to the call that asked for it.

View Source 
const DefaultExecutionBudget int32 = 10_000

DefaultExecutionBudget is how many element activations one token may drive in a single run before the engine parks it with an incident. Ten thousand is far above any plausible stretch of automatic work for one token — a token that has taken ten thousand sequence flows without once waiting for a job, a timer or a message is not making progress a person modelled — and low enough that the writer comes back in well under a second.

View Source 
const DefaultMaxCollection int64 = 16 << 20

DefaultMaxCollection is how large a multi-instance activity's output collection may be. It has to clear a legitimate loop at the iteration ceiling — a hundred thousand modest results — while staying far below what costs the host its memory.

View Source 
const DefaultMaxIterations = 100_000

DefaultMaxIterations is how many iterations one multi-instance activity may ask for before the engine refuses it with an incident.

It is a *size* budget where DefaultExecutionBudget is a *rate* one, and neither substitutes for the other: a hundred thousand iterations are a hundred thousand tokens taking one step each, which the execution budget is deliberately built not to stop. What makes them dangerous is that the count comes from the model or from an instance variable, and the engine allocated from it before looking — a variable holding a billion is a billion FEEL nulls, asked for in one call (ADR-0276).

A hundred thousand is far above what a modelled loop plausibly wants and far below the point where the allocation is the problem.

View Source 
const DefaultMaxVariable int64 = 1 << 20

DefaultMaxVariable is how large one variable's value may be. A megabyte holds a business record with room to spare; past that it is a document, and a document in a token's scope is rewritten into the log on every touch.

View Source 
const LoopCounterVariable = compiler.LoopCounterVariable

LoopCounterVariable is the standard multi-instance per-iteration counter variable (1-based), bound into each inner iteration's scope (ADR-0077, matching Zeebe). Exported so read-side surfaces can name the variable they are looking for without re-inventing the string and drifting from what the engine actually writes. It is the compiler's constant, so the deploy check that keeps a model from mapping onto it (loop.counter-mapping) and the runtime that writes it can never disagree.

Variables

View Source 
var BuildVersion string

BuildVersion is the Atlas build recorded in checkpoint manifests for diagnostics (ADR-0131). It is metadata only — nothing branches on it — and the server may stamp it at startup; an unset value simply records no version.

Functions

This section is empty.

Types

type Arrival added in v0.6.0 

type Arrival struct {
	Key  uint64
	Flow int32
}

Arrival is one token waiting on a join: the element instance holding it and the incoming sequence flow it came in on.

The flow is what a join has to count by. BPMN 2.0.2 §13.4 activates a parallel gateway when there is at least one token on *each* incoming sequence flow, and consumes exactly one from each — which is a different question from how many tokens are on the node, and gives a different answer whenever a flow carries two (ADR-0290).

type BatchStats added in v0.2.0 

type BatchStats struct {
	// Commands is how many queued commands the batch consumed.
	Commands int
	// Events is how many events it made durable — zero for a batch whose commands
	// produced nothing to write.
	Events int
	// QueueDepth is how many commands remain queued after the batch, including the
	// follow-ups it scheduled. It is the backpressure signal: a depth that climbs
	// batch over batch means the writer is not keeping up with what is arriving.
	QueueDepth int
	// SyncSeconds is how long the batch's single group-commit fsync took.
	SyncSeconds float64
	// CommitSeconds is how long making the batch's state visible took.
	CommitSeconds float64
	// Jobs is what the batch did to the job lifecycle (ADR-0142 slice 5).
	Jobs JobStats
}

BatchStats is what one durably committed batch did. It is passed by value and never retained by the engine.

SyncSeconds and CommitSeconds are zero for a batch that produced no events: there was nothing to fsync and nothing to commit, so an implementation should record no observation rather than a zero one, which would drag a latency histogram down.

type CallTargetOverride 

type CallTargetOverride struct {
	// Disabled parks the call (as an undeployed callee does) instead of resolving.
	Disabled bool
	// PinnedDefKey resolves to exactly this definition key (0 = not pinned). The
	// server layer picks the key from an operator-named version; the engine is
	// version-agnostic and simply uses it, parking if it is no longer deployed.
	PinnedDefKey uint64
	// RedirectProcessId resolves the newest deployment of this process id instead of
	// the called one ("" = no redirect). A redirect uses the default `latest`
	// resolution for its target (no chaining), so overrides cannot form a cycle.
	RedirectProcessId string
}

CallTargetOverride redirects, pins, or disables a call activity's target on this server (ADR-0105). Exactly one shape is meaningful per record; the resolution precedence (see ProcessingContext.resolveCallTarget) is Disabled, then PinnedDefKey, then RedirectProcessId, else the default `latest` resolution. It is operator config, not derived from deployments and not event-sourced — it changes only future resolutions; a child already created carries its frozen def key, so replay is unaffected (I6).

type Clock 

type Clock interface {
	Now() int64 // unix nanoseconds
}

Clock supplies wall-clock time. It is injected so tests can drive time deterministically (invariant I4: time is read into events, never inside applyToState).

type Command 

type Command struct {
	Key       uint64
	ValueType model.ValueType
	Intent    model.Intent
	Value     inflightValue
	// SourcePos is the log position of the event that scheduled this command
	// (0 for externally submitted commands), used to thread causality into the
	// events the command produces.
	SourcePos uint64
	// StartVars carries variables attached to a command: the initial variables for
	// a process-instance creation command, or the output variables a worker
	// produced for a job-completion command. Both are external, non-hot-path
	// intents, so a slice here does not affect the token-movement fast path.
	StartVars []model.VariableValue
	// StartElements names the root start events an instance-creation command seeds a
	// token at. A *triggered* start — a correlating message, a broadcast signal, a
	// fired start timer — carries exactly the one that fired, because a start event is
	// a trigger and the one that happened is the one that instantiates.
	//
	// nil is the untriggered create (the API, a call activity), which seeds the
	// process's own entry points instead; see startElementsFor. The zero value is
	// therefore the safe one: a command that forgets to say what triggered it starts
	// the process the way pressing Start does, never at somebody else's trigger.
	//
	// It rides only on the non-hot-path creation intent, alongside StartVars, so it
	// never touches token movement (invariant I1).
	StartElements []int32
	// Decision carries a DMN decision evaluation a worker produced for a
	// job-completion command (ADR-0066): the inputs, outputs, and trace it froze off
	// the processor goroutine, recorded as history when the completion is folded. It
	// is nil for every other command and for job completions that are not decisions,
	// so — like StartVars, on the same non-hot-path completion intent — it never
	// touches the token-movement fast path.
	Decision *model.DecisionEvaluationValue
	// ToolCalls carries the tools an agent chose for the next round of an agent-driven
	// ad-hoc subprocess (ADR-0253), riding on that container's job-completion command the
	// way Decision rides on a business rule task's. It is engine control data, not process
	// data: putting it in StartVars would leak it into FEEL, the variable timeline and
	// every downstream expression. Empty for every other command — and an empty list on
	// an agent container's completion is the agent saying it is done, which is why the
	// zero value is the ending, not an error.
	ToolCalls []model.ToolCall
	// Actor identifies who submitted an external variable-modify command (ADR-0098):
	// the acting principal's username, frozen into the audit event the modify emits so
	// the "who changed it" trail is durable and replayable. Empty for every other
	// command (and for a modify made with auth off / by an unidentified caller). It
	// rides only on the non-hot-path IntentVariableModify command, so it never touches
	// the token-movement fast path.
	Actor string
	// Reason is the operator's justification for an intervention that forces a step the
	// engine would not have taken on its own — completing a parked job by hand (ADR-0159).
	// It rides only on such a command, alongside Manual, and is frozen into the audit event
	// the handler emits, so "why" is as durable and replayable as "who". Empty for every
	// other command, so it never touches the token-movement fast path.
	Reason string
	// Manual marks a completion an operator forced rather than a worker reporting real
	// work (ADR-0159). It is the explicit gate for the audit record — never inferred from
	// Reason being set — so a manual completion is always attributable even if the reason
	// is somehow empty, and a worker's completion never mints an operator-action record.
	Manual bool
	// RetryBackoff is the delay (unix-nanoseconds) a worker asked to wait before its failed
	// job may be retried (ADR-0111). It rides only on the non-hot-path IntentJobFailed command;
	// the handler reads the clock at command time and freezes now+RetryBackoff into the job's
	// RetryDueDate (invariant I6). 0 means retry immediately (the pre-0111 behavior).
	RetryBackoff int64
	// LeaseFor is how long (in nanoseconds) a worker asked to hold a job it is activating
	// (ADR-0007). It rides only on the non-hot-path IntentJobActivated command; like
	// RetryBackoff, the handler reads the clock at command time and freezes
	// now+LeaseFor into the job's LeaseExpiresAt, so a replay lands on the same instant
	// (invariant I6).
	LeaseFor int64
}

Command is an intention handed to the processor. Commands are processed but never persisted (only the events they produce are); on recovery they are not replayed (invariant I6). The payload is held by value (see inflightValue) so queuing a command does not allocate.

type JobStats added in v0.2.0 

type JobStats struct {
	// Created counts jobs that became available to a worker.
	Created int
	// Completed counts jobs a worker finished successfully.
	Completed int
	// Failed counts worker-reported failures. A failure with retries left leaves the
	// job open for another attempt; one without parks it with an incident (ADR-0061).
	Failed int
	// Canceled counts jobs removed without being worked — their element was
	// interrupted, terminated, or its instance cancelled.
	Canceled int
}

JobStats counts the job-lifecycle transitions one batch made durable. It rides on BatchStats rather than a separate call so it inherits the same durability ordering: a job is counted as created only once the event that created it is on disk.

The lease-based worker protocol (ADR-0007) is not built yet, so activations, lease expiries and timeouts have no events to count and are absent rather than reported as a permanent zero — a zero timeout counter on an engine that cannot time out reads as "nothing is timing out", which is true but misleading.

type Metrics added in v0.2.0 

type Metrics interface {
	// BatchCommitted reports a batch whose events are durable and whose state is
	// committed. It is never called for a batch that failed.
	BatchCommitted(BatchStats)
	// SyncFailed reports a batch whose group-commit fsync failed. Nothing it wrote is
	// durable.
	SyncFailed()
	// CommitFailed reports a batch whose events are durable but whose state commit
	// failed. Recovery will re-apply them from the log.
	CommitFailed()
}

Metrics observes the batch loop. Every method is called from the single-writer goroutine (invariant I3), so an implementation must not block, must not call back into the processor, and must not allocate (invariant I1).

A nil Metrics — the default — means the engine reports nothing and does not even read the clock, so an uninstrumented processor pays literally nothing.

type MigrationElement added in v0.3.0 

type MigrationElement struct {
	Key             uint64
	ElementId       int32
	FlowScopeKey    uint64
	BpmnElementType uint8
	MultiInstance   uint8
	EventGatewayKey uint64
	AttachedToKey   uint64
}

MigrationElement is one live element instance, as the migration validator needs to see it. It is a plain struct rather than model.ElementInstanceValue so the validator can be called from the API — which reads the store — and from the processor — which reads the in-flight transaction — over the same rules (ADR-0162).

func MigrationElementOf added in v0.3.0 

func MigrationElementOf(key uint64, ei *model.ElementInstanceValue) MigrationElement

MigrationElementOf projects a stored element instance onto what the validator reads.

type MigrationProblem added in v0.3.0 

type MigrationProblem struct {
	ElementInstanceKey uint64 `json:"elementInstanceKey"`
	ElementID          string `json:"elementId"`
	Reason             string `json:"reason"`
}

MigrationProblem is one reason a migration is refused, named so an operator can act on it: which element instance, which element the modeler would recognise, and what is wrong with moving it.

func ValidateMigration added in v0.3.0 

func ValidateMigration(from, to *compiler.CompiledProcess, live []MigrationElement, mapping map[int32]int32) []MigrationProblem

ValidateMigration reports every reason the given mapping cannot rebind these live element instances from the `from` definition to the `to` one. An empty result means the migration is safe to submit; anything else refuses it (ADR-0162).

It refuses rather than guesses, and that is the whole point of it. A token left on an index that means something else in the target graph, or landed on an element of another type, corrupts an instance in a way no later fix repairs — while a refused migration costs an operator one message. So every rule here answers "could this rebinding produce an instance the engine can still run correctly?", and anything it cannot answer yes to is a problem.

It is a pure function of its arguments: the API calls it to refuse before submitting a command, and the processor calls it again on the run loop, where the state may have moved since. It never runs inside applyToState — the fold reads the mapping the event froze and computes nothing (invariant I4).

type ProcessingContext 

type ProcessingContext struct {
	// contains filtered or unexported fields
}

ProcessingContext is the surface every behavior works through while a command is processed. A behavior may do three things: read state, write events (a fact that also mutates state), and schedule what comes next. It never touches the log or fsync directly — it only accumulates into the batch (invariant I2: nothing becomes visible before the batch is durable).

func (*ProcessingContext) ActiveChildren 

func (c *ProcessingContext) ActiveChildren(scope uint64) int32

ActiveChildren returns the active-child count of a scope (e.g. to detect that a process instance has finished).

func (*ProcessingContext) AppendCompensableEvent 

func (c *ProcessingContext) AppendCompensableEvent(intent model.Intent, v model.CompensableValue)

AppendCompensableEvent records a compensation-index change: IntentCompensableRecorded retains a completed compensable activity (keyed under its scope in completion order), and IntentCompensableConsumed drops one once it has been compensated (ADR-0103). Both ride only on the command path (a completion or a compensation throw), never token movement; applyToState folds them into the compensable index so recovery rebuilds it (invariant I6). Keyed by the owning process instance, like the other history events.

func (*ProcessingContext) AppendCreateChildInstanceCommand 

func (c *ProcessingContext) AppendCreateChildInstanceCommand(defKey uint64, vars []model.VariableValue, parentElementKey uint64)

AppendCreateChildInstanceCommand is AppendCreateInstanceCommand for a call activity: the created instance records the caller's call-activity element instance as its parent, so on completion it resumes that element (ADR-0076).

func (*ProcessingContext) AppendCreateInstanceCommand 

func (c *ProcessingContext) AppendCreateInstanceCommand(defKey uint64, vars []model.VariableValue, correlationKey string, startElement int32)

AppendCreateInstanceCommand schedules creation of a new instance of defKey for a later batch, seeded with vars (each re-scoped to the new instance when it is created) and the correlationKey the created instance records (empty for a timer or signal start). A correlating message uses it to instantiate a message-start process (ADR-0035). Deferring to a followup keeps instance creation on the same command path as an API-submitted create, so its events — and thus recovery — are identical however the create was triggered.

startElement is the root start event that fired, and every caller here has one: a trigger instantiates at itself, not at every entry the process happens to have (ADR-0226). It is a required argument rather than an optional one so that adding a fourth kind of trigger cannot quietly inherit the old seed-everything behaviour.

func (*ProcessingContext) AppendDataObjectEvent 

func (c *ProcessingContext) AppendDataObjectEvent(intent model.Intent, v model.DataObjectValue)

AppendDataObjectEvent records a data-object write (created or state-changed). Like a variable it carries genuine runtime data (a name, a data state, and a value), so it allocates for its strings — data objects are runtime data, not hot-path token movement (ADR-0053). The event is keyed by the owning scope.

func (*ProcessingContext) AppendDecisionEvaluationEvent 

func (c *ProcessingContext) AppendDecisionEvaluationEvent(v model.DecisionEvaluationValue)

AppendDecisionEvaluationEvent records how a business rule task's decision was made — its inputs, outputs, and trace — as append-only history (ADR-0066). The worker evaluated the decision off the processor goroutine and froze the result onto the completion command; this event carries genuine runtime data (JSON payloads), so it allocates for its strings, not hot-path token movement. It is keyed by the owning process instance, so a scope-wide scan yields every decision an instance evaluated in order.

func (*ProcessingContext) AppendElementCommand 

func (c *ProcessingContext) AppendElementCommand(key uint64, intent model.Intent, v model.ElementInstanceValue)

AppendElementCommand schedules an element-instance command for a later batch.

func (*ProcessingContext) AppendElementEvent 

func (c *ProcessingContext) AppendElementEvent(key uint64, intent model.Intent, v model.ElementInstanceValue)

AppendElementEvent records an element-instance lifecycle fact.

func (*ProcessingContext) AppendInboundDeliveryEvent 

func (c *ProcessingContext) AppendInboundDeliveryEvent(v model.InboundDeliveryValue)

AppendInboundDeliveryEvent advances an external source's inbound high-water mark (ADR-0075), keyed on the receiving definition space as a neutral key (the record carries the source id and sequence it needs). Emitted in the same batch as the message publish it guards, so the dedup mark and the effects it authorizes commit atomically under one fsync (invariant I2).

func (*ProcessingContext) AppendIncidentEvent 

func (c *ProcessingContext) AppendIncidentEvent(intent model.Intent, v model.IncidentValue)

AppendIncidentEvent records an incident lifecycle fact (created or resolved). The key is the element instance the incident is attached to, and the value carries that key too, so applyToState can locate the index entry from the event alone on either intent (ADR-0061).

func (*ProcessingContext) AppendJobEvent 

func (c *ProcessingContext) AppendJobEvent(key uint64, intent model.Intent, v model.JobValue)

AppendJobEvent records a job lifecycle fact.

func (*ProcessingContext) AppendMessageFlowEvent 

func (c *ProcessingContext) AppendMessageFlowEvent(v model.MessageFlowValue)

AppendMessageFlowEvent retains one delivered message flow as history for the collaboration replay (ADR-0038). It is keyed by its receiving definition (the state index leads with it); the event's header timestamp and position order it on the replay timeline. Emitted once per correlated catch event and once per message-start instantiation, so both kinds of cross-pool delivery are recorded.

func (*ProcessingContext) AppendMessageSubscriptionEvent 

func (c *ProcessingContext) AppendMessageSubscriptionEvent(key uint64, intent model.Intent, v model.MessageSubscriptionValue)

AppendMessageSubscriptionEvent records a message-subscription fact (created or correlated). The key is the waiting element instance's key, and the value carries the match pair, so applyToState can locate the index entry from the event alone (invariant I4).

func (*ProcessingContext) AppendMigrationEvent added in v0.3.0 

func (c *ProcessingContext) AppendMigrationEvent(v model.ProcessMigrationValue)

AppendMigrationEvent rebinds a running instance to another deployed version of its process (ADR-0162). Unlike every other event it describes no single entity's transition: it carries the whole element mapping the fold rewrites the instance's live records through, frozen at command time so replay reproduces the rebinding from the log rather than by re-deriving it (invariants I4/I6).

func (*ProcessingContext) AppendOperatorActionEvent added in v0.3.0 

func (c *ProcessingContext) AppendOperatorActionEvent(v model.OperatorActionValue)

AppendOperatorActionEvent records that an operator intervened on a running instance — completing a parked job by hand, say (ADR-0159). Like AppendVariableAuditEvent it is pure history: emitted alongside the events the intervention produces, it freezes who acted and why into the log, so a forced step is never indistinguishable from one the engine drove and replay rebuilds the identical trail (invariant I6).

func (*ProcessingContext) AppendProcessInstanceCommand 

func (c *ProcessingContext) AppendProcessInstanceCommand(key uint64, intent model.Intent, v model.ProcessInstanceValue)

AppendProcessInstanceCommand schedules a process-instance command (e.g. the Terminating that cancels a call activity's child) for a later batch — the same command an API cancel enqueues, so the child tears down through the identical path however its termination was triggered (ADR-0076).

func (*ProcessingContext) AppendProcessInstanceEvent 

func (c *ProcessingContext) AppendProcessInstanceEvent(key uint64, intent model.Intent, v model.ProcessInstanceValue)

AppendProcessInstanceEvent records a process-instance lifecycle fact.

func (*ProcessingContext) AppendSignalSubscriptionEvent 

func (c *ProcessingContext) AppendSignalSubscriptionEvent(key uint64, intent model.Intent, v model.SignalSubscriptionValue)

AppendSignalSubscriptionEvent records a signal-subscription fact (created or correlated). The key is the waiting element instance's key, and the value carries the signal name, so applyToState can locate the index entry from the event alone (invariant I4). A signal reuses the message subscription intents (SubscriptionCreated / SubscriptionCorrelated) over a separate family (ADR-0088).

func (*ProcessingContext) AppendTimerEvent 

func (c *ProcessingContext) AppendTimerEvent(key uint64, intent model.Intent, v model.TimerValue)

AppendTimerEvent records a timer lifecycle fact (created or triggered).

func (*ProcessingContext) AppendVariableAuditEvent 

func (c *ProcessingContext) AppendVariableAuditEvent(v model.VariableAuditValue)

AppendVariableAuditEvent records who set a variable from outside the model — an operator override — as append-only audit history (ADR-0098). Like a variable it carries genuine runtime data (an actor, a name, and contents), so it allocates for its strings; it rides only on the non-hot-path variable-modify command, never token movement. It is keyed by the owning process instance, so a scope-wide scan yields every override an instance received in order.

func (*ProcessingContext) AppendVariableEvent 

func (c *ProcessingContext) AppendVariableEvent(intent model.Intent, v model.VariableValue) bool

AppendVariableEvent records a variable write and reports whether it happened: a value past the variable budget is refused, with an incident naming it (ADR-0294). Most callers write engine-derived values — a loop index, a counter — which cannot exceed a budget sized for a business record, and they ignore the result. A caller that writes something a model or a worker produced should not.

The value is data (a name and contents), so unlike the graph-derived events this one does allocate for its strings — variables are runtime data, not hot-path token movement.

func (*ProcessingContext) ArrivalsOnNode added in v0.6.0 

func (c *ProcessingContext) ArrivalsOnNode(procKey, scopeKey uint64, elementId int32) []Arrival

ArrivalsOnNode returns the tokens waiting on the given node in one execution scope, oldest first, each with the flow it arrived on. The order is the index's: element-instance keys ascend with their minting, so oldest-first is what the scan yields, and it is what makes a join's choice of which token to consume both first-in-first-out and a pure function of state.

The result aliases a processor-owned buffer and is valid until the next call.

func (*ProcessingContext) ChildInstancesOf added in v0.5.0 

func (c *ProcessingContext) ChildInstancesOf(callElKey uint64) []uint64

ChildInstancesOf returns the live child process instances a call-activity element instance started, from the committed childByParent index (ADR-0076).

It replaced a walk of every live process instance comparing each one's ParentElementInstanceKey. That walk was O(instances) *per call activity torn down*, so cancelling a parent holding many children cost the product of the two — on the single-writer loop, which is every other request's queue as well. The keys are collected before the caller acts on them, as the walk did, so the caller may emit events for each without disturbing the read.

Read through the transaction, so a child activated earlier in this same batch is visible to the teardown. ADR-0238 originally read the committed store here and called that a determinism measure; it is not one. The records this transaction has already applied are a deterministic function of the commands processed so far — replay applies the same records in the same order — so seeing them is as reproducible as not seeing them, and strictly more correct. What the committed view actually produced was a child that outlived the caller cancelled in the batch that created it (ADR-0284).

func (*ProcessingContext) ForEachElementInstance 

func (c *ProcessingContext) ForEachElementInstance(procKey uint64, fn func(elKey uint64))

ForEachElementInstance calls fn with the key of every element instance belonging to a process instance, through the in-flight transaction — so it sees element instances created earlier in the same batch, consistently with GetElementInstance, GetJob, and ActiveChildren (all tx-reads). This matters for a terminate end event reached in the same batch as a parallel sibling's activation (ADR-0116): the sibling is not yet committed, but it is in the tx, so the scope teardown finds it. Keys are collected before fn runs so fn may mutate element-instance state (e.g. emit terminations) without disturbing the scan.

func (*ProcessingContext) ForEachStartTimer 

func (c *ProcessingContext) ForEachStartTimer(fn func(key uint64, v model.TimerValue))

ForEachStartTimer calls fn with the key and value of every armed start timer, read from the committed timer index. Entries are collected before fn runs so fn may emit timer events (arming/retiring) without disturbing the scan. Used only when a definition is deployed (off the hot path), to arm and supersede start timers (ADR-0051).

func (*ProcessingContext) GetDataObject 

func (c *ProcessingContext) GetDataObject(scope uint64, name string) *model.DataObjectValue

GetDataObject reads a scope's data object by name through the in-flight transaction (sees writes from earlier in this batch). A data-output association uses it to keep the object's current value or state when the write changes only one of them (ADR-0058); nil if the object is absent.

func (*ProcessingContext) GetElementInstance 

func (c *ProcessingContext) GetElementInstance(key uint64) *model.ElementInstanceValue

GetElementInstance reads element-instance state through the in-flight transaction (sees this batch's uncommitted writes).

func (*ProcessingContext) GetIncident 

func (c *ProcessingContext) GetIncident(elKey uint64) *model.IncidentValue

GetIncident reads the incident attached to an element instance through the in-flight transaction, or nil if there is none (ADR-0061).

func (*ProcessingContext) GetJob 

func (c *ProcessingContext) GetJob(key uint64) *model.JobValue

GetJob reads job state through the in-flight transaction.

func (*ProcessingContext) GetProcessInstance 

func (c *ProcessingContext) GetProcessInstance(key uint64) *model.ProcessInstanceValue

GetProcessInstance reads process-instance state through the in-flight transaction.

func (*ProcessingContext) GetVariable 

func (c *ProcessingContext) GetVariable(scope uint64, name string) *model.VariableValue

GetVariable reads a scope's variable by name through the in-flight transaction (sees writes from earlier in this batch, e.g. seeded start variables).

func (*ProcessingContext) IsCanceling 

func (c *ProcessingContext) IsCanceling(txKey uint64) bool

IsCanceling reports whether the transaction scope txKey was marked cancelling by a cancel end event (ADR-0108).

func (*ProcessingContext) JobOfElement 

func (c *ProcessingContext) JobOfElement(elKey uint64) (uint64, bool)

JobOfElement returns the key of the job held by an element instance and whether it holds one, through the in-flight transaction. An interrupting boundary event uses it to cancel the host activity's job when it terminates the host.

func (*ProcessingContext) NewKey 

func (c *ProcessingContext) NewKey() uint64

NewKey mints a fresh entity key. The minted key is frozen into the event that uses it, so replay reproduces it without regeneration (invariant I6).

func (*ProcessingContext) NotifyJobAvailable 

func (c *ProcessingContext) NotifyJobAvailable(jobType int32)

NotifyJobAvailable registers a post-fsync notification that a job of the given type is available (invariant I2: runs after the batch is durable).

func (*ProcessingContext) Now 

func (c *ProcessingContext) Now() int64

Now reads wall-clock time. It is captured into events here, never inside applyToState (invariant I4).

func (*ProcessingContext) OldestPerFlow added in v0.6.0 

func (c *ProcessingContext) OldestPerFlow(arrivals []Arrival) []Arrival

OldestPerFlow reduces arrivals to one token per distinct incoming flow — the oldest on each — which is the set a join firing consumes. Everything it leaves out is surplus: a second token on a flow that already has one, which belongs to the *next* firing and must not be swallowed by this one.

The result aliases a second processor-owned buffer, so it stays valid across a re-scan of the node.

func (*ProcessingContext) ResolveVariable 

func (c *ProcessingContext) ResolveVariable(startScope uint64, name string) *model.VariableValue

ResolveVariable reads name resolving up the scope chain from startScope (nearest scope wins), the lookup activity-local scopes and Camunda-style I/O mappings need (ADR-0068). A scope's parent is its element instance's FlowScopeKey; the root process-instance scope has no element instance, which ends the walk. Reads go through the in-flight transaction, so they see this batch's writes.

func (*ProcessingContext) TokenCanStillReach 

func (c *ProcessingContext) TokenCanStillReach(procKey, scopeKey uint64, nodeId int32, reaches compiler.NodeSet) bool

TokenCanStillReach reports whether any live token in scopeKey could still arrive at nodeId: an active element instance sitting on a node from which nodeId is reachable (per reaches), or a token in flight as an element-activating command not yet processed — the rest of this batch's queue plus followups generated so far — targeting such a node or nodeId itself. Tokens already parked on nodeId are the join's own arrivals and are excluded. An inclusive join fires only when this is false. Considering in-flight commands is what keeps two pass-through branches from each firing the join separately.

scopeKey scopes the question the same way it scopes a parallel join's count (ADR-0277): a token in a *sibling* iteration of a multi-instance subprocess is on the same node ids but can never arrive here, and letting it hold the join open made every iteration wait for the slowest one.

Narrowing to one scope does not blind the join to a token running inside a nested subprocess on one of its branches: reaches follows sequence flows, which never cross a scope boundary, so an inner node is not in reaches at all — what keeps the join waiting is the subprocess's own element instance, which sits in *this* scope on a node that does reach the join.

func (*ProcessingContext) VariablesOfScope 

func (c *ProcessingContext) VariablesOfScope(scope uint64, fn func(v model.VariableValue))

VariablesOfScope calls fn with each variable owned by scope, read through the in-flight transaction (so it sees this batch's writes). Values are collected into a fresh slice before fn runs, so fn may emit variable events (e.g. deleting the scope's locals) without disturbing the underlying scan. Used to drop an activity-local scope on completion (ADR-0068).

type Processor 

type Processor struct {
	// contains filtered or unexported fields
}

Processor owns one partition's command processing.

func New 

func New(partition uint16, log *wal.Log, store *state.Store, clock Clock) *Processor

New creates a processor for the given partition over an open log and store. A nil clock defaults to the system clock.

func (*Processor) ActivateJob added in v0.3.0 

func (p *Processor) ActivateJob(jobKey uint64, worker string, leaseFor int64)

ActivateJob enqueues an external worker's claim on a job (ADR-0007): the job is held off the activatable index for leaseFor nanoseconds and recorded as held by worker, so no other worker is offered it while this one works.

The lease is a bound, not a lock. When it elapses the job is offered again — that is what makes a worker crash recoverable without an operator — so a worker that outlives its lease may find its work has been handed on. Activating a job that is gone, or one another worker already holds, is a no-op. Call RunUntilIdle (or Drive) to process it.

func (*Processor) ArmStartTimers 

func (p *Processor) ArmStartTimers(defKey uint64)

ArmStartTimers enqueues arming of a freshly deployed definition's timer start events: the handler creates their durable timers and retires any that a prior version of the same process left armed, so only the latest version's schedule is active (ADR-0051). Call it once per *fresh* deploy (not on recovery — the restored TimerCreated events already hold the armed timers), then RunUntilIdle (or Drive) to process it. It scans the armed start timers, so callers skip it for a first-version process with no timer start events (nothing to arm or supersede); a re-version still calls it so a removed schedule is retired.

func (*Processor) AssignJob 

func (p *Processor) AssignJob(jobKey uint64, assignee string)

AssignJob enqueues a (re)assignment of a user task's assignee, identified by its job key. A non-empty assignee is a claim; an empty one unclaims the task, making it available again. The job stays open either way. Assigning a job that no longer exists is a no-op. Call RunUntilIdle to process it (ADR-0042).

func (*Processor) CancelInstance 

func (p *Processor) CancelInstance(piKey uint64)

CancelInstance enqueues termination of a running process instance: every active element instance is terminated and the instance is recorded as terminated in history (ADR-0017). Any timer/subscription/job the instance left waiting is self-retiring — when it later fires or correlates it finds no element and does nothing. Call RunUntilIdle to process it.

func (*Processor) Checkpoint added in v0.2.0 

func (p *Processor) Checkpoint(root string) (uint64, error)

Checkpoint writes a recovery checkpoint of the currently applied state under root and returns the applied log position it captures (ADR-0131).

It **must be called on the partition's single-writer goroutine, between batches** (invariant I3). That is what makes the checkpoint's consistency boundary exact: no state mutation can race the snapshot, so the position it records is precisely the position the snapshotted state contains. Calling it concurrently with the run loop would produce a snapshot at a fuzzy position — the failure ADR-0131 rejects.

It is purely additive to durability (invariant I2): a checkpoint is an optimization that lets a later recovery replay only the WAL suffix past its applied position. Nothing here writes to the log, mutates state, or acknowledges anything, so a failed or absent checkpoint costs only a slower recovery, never correctness — the caller may log the error and retry on the next cadence.

Restoring from a checkpoint, and deleting the WAL segments it makes redundant, are the later ADR-0131 slices; this only produces one.

func (*Processor) ClearCallTargetOverride 

func (p *Processor) ClearCallTargetOverride(calledProcessId string)

ClearCallTargetOverride removes a called process id's override, restoring the default `latest` resolution. Idempotent. Run-loop goroutine only.

func (*Processor) CompactLog added in v0.2.0 

func (p *Processor) CompactLog(checkpointRoot string, consumerLimits []uint64) (int, error)

CompactLog deletes the WAL segments no longer needed for recovery and returns how many were removed (ADR-0131). It must be called on the single-writer goroutine.

A segment is removed only when it lies entirely at or below the **compaction cut**, which is the minimum of:

  • the newest checkpoint for this partition that **fully verifies** — manifest *and* state files — and whose applied position is at or below the store's. Verification is stricter here than in RecoverFrom on purpose: once the prefix is deleted, that checkpoint's state files become the only way to rebuild it, so a corrupt snapshot must never license a deletion. Requiring it to be at or below the store's applied position matters just as much: recovery refuses a checkpoint ahead of the store (it skips reading a prefix, it does not install state files), so deleting below such a checkpoint would strand recovery with a gap it can no longer replay.
  • every consumerLimit the caller passes — the exported-log high-water mark (ADR-0114) and the retention safe position (ADR-0115) when those are enabled. The caller owns this list because only it knows which consumers exist; passing nil means "no consumers", so pass every enabled consumer's watermark or risk deleting records it has not read yet.

If no checkpoint qualifies the cut is zero and **nothing is deleted** — the log stays the sole recovery source rather than being trimmed on a promise that cannot be checked. Compaction is an optimization like the checkpoint itself: skipping it costs disk, never correctness.

func (*Processor) CompleteJob 

func (p *Processor) CompleteJob(jobKey uint64, outputs ...model.VariableValue)

CompleteJob enqueues completion of a job by a worker, optionally carrying the output variables the worker produced (e.g. a business rule task's decision result). The outputs are written into the job's process instance scope when the completion is processed, before the element completes, so a downstream gateway can route on them. They are frozen into VariableCreated events, so replay re-applies them without re-running the worker (invariant I6).

func (*Processor) CompleteJobManually added in v0.3.0 

func (p *Processor) CompleteJobManually(jobKey uint64, actor, reason string, outputs ...model.VariableValue)

CompleteJobManually completes a job the way a worker would, but on an operator's say-so rather than because the work was reported (ADR-0159): the job is completed exactly as CompleteJob does, and the intervention is additionally frozen into an append-only audit event carrying who forced it and why. actor is the acting principal's username ("" when auth is off); reason is their justification. Use this, never CompleteJob, for anything a person triggers, so a forced step is never indistinguishable from one the engine drove.

func (*Processor) CompleteJobWithDecision 

func (p *Processor) CompleteJobWithDecision(jobKey uint64, decision *model.DecisionEvaluationValue, outputs ...model.VariableValue)

CompleteJobWithDecision completes a business rule task's job like CompleteJob, additionally carrying the DMN decision evaluation the worker produced (ADR-0066): its inputs, outputs, and trace, frozen into a history event when the completion is folded so an operator can later inspect how the decision was made. decision may be nil, in which case this behaves exactly like CompleteJob.

func (*Processor) CompleteJobWithToolCalls added in v0.5.0 

func (p *Processor) CompleteJobWithToolCalls(jobKey uint64, toolCalls []model.ToolCall, outputs ...model.VariableValue)

CompleteJobWithToolCalls completes an agent-driven ad-hoc subprocess's round job (ADR-0253), carrying the tools the model chose to run next. An empty or nil toolCalls is the agent reporting it is done: the container then completes through the ordinary path, exactly as CompleteJob would drive it. Any other job type ignores the calls, so this is only meaningful on a container's round job.

func (*Processor) CreateInstance 

func (p *Processor) CreateInstance(defKey uint64, startVars ...model.VariableValue)

CreateInstance enqueues creation of a new instance of the given definition, optionally seeded with initial variables. Call RunUntilIdle to process it.

func (*Processor) Deploy 

func (p *Processor) Deploy(cp *compiler.CompiledProcess)

func (*Processor) FailJob 

func (p *Processor) FailJob(jobKey uint64, retries int32, message string, backoff int64)

FailJob enqueues a worker's failure report for a job (ADR-0061), carrying the retries the worker leaves it, a failure message, and a retry backoff (unix-nanoseconds; 0 = retry immediately, ADR-0111). With retries > 0 the job is retried — immediately if backoff is 0, otherwise held off the activatable index until a retry timer fires backoff nanoseconds later; with retries <= 0 an incident is raised on the job's element and the token parks there. Failing a job that no longer exists is a no-op. Call RunUntilIdle (or Drive) to process it.

func (*Processor) LastRecovery added in v0.2.0 

func (p *Processor) LastRecovery() RecoveryStats

LastRecovery returns what the last Recover/RecoverFrom did.

It is read at scrape time rather than pushed, which is what keeps it simple: recovery happens once, before the server that would hold a metrics registry exists, so a pushed counter would have nowhere to go. The fields are written by the goroutine that runs recovery and read only after it returns, so the construction that follows establishes the happens-before (invariant I3 is untouched — this never reaches partition state).

func (*Processor) MigrateInstance added in v0.3.0 

func (p *Processor) MigrateInstance(v model.ProcessMigrationValue, actor, reason string)

MigrateInstance enqueues the rebinding of a running instance from one deployed version of its process to another (ADR-0162), carrying the fully materialized element mapping the fold will rewrite its live records through, plus who asked and why.

The mapping is built and validated by the caller — it needs both compiled processes and the deployment records, which the API holds — and re-checked here on the run loop before anything is emitted, because the instance is free to move in between. A migration that no longer holds is dropped; the caller's refusal is the one an operator reads. Call RunUntilIdle to process it.

func (*Processor) Partition added in v0.5.0 

func (p *Processor) Partition() uint16

Deploy registers an immutable compiled definition so instances can run it, and indexes any message start events so a correlating message instantiates it (ADR-0035). Partition reports which partition this processor drives. It is read-only and fixed for the processor's life, so it is safe to call from any goroutine — the API's node descriptor (ADR-0189 §6) reads it while serving a request, off the run loop.

func (*Processor) ProcessActive added in v0.2.0 

func (p *Processor) ProcessActive(defKey uint64) bool

ProcessActive reports whether a definition may auto-start new instances — the inverse of the ADR-0119 deactivation flag. A key that was never deactivated (the default) is active. Run-loop goroutine only.

func (*Processor) PublishInbound 

func (p *Processor) PublishInbound(sourceID string, seq uint64, name, correlationKey string, vars ...model.VariableValue)

PublishInbound enqueues publication of a message that originated from an external event source (ADR-0075), carrying the source's identity (sourceID) and monotonic sequence (seq) so the publish is deduplicated against the source's durable high-water mark: a replayed at-least-once delivery is skipped rather than re-correlated (which would double-start a message-start process). Apart from the dedup guard it correlates exactly like PublishMessage. Call RunUntilIdle to process it.

func (*Processor) PublishMessage 

func (p *Processor) PublishMessage(name, correlationKey string, vars ...model.VariableValue)

PublishMessage enqueues publication of a message with the given name and correlation key, optionally carrying payload variables that are written into every correlated instance's scope. It correlates against open subscriptions through the same path a message throw event uses; a message that matches no subscription is a no-op (no buffering yet, ADR-0020). Call RunUntilIdle to process it.

func (*Processor) PurgeInstance added in v0.2.0 

func (p *Processor) PurgeInstance(piKey uint64, pi *model.ProcessInstanceValue)

PurgeInstance enqueues the hard delete of a finished instance's history (ADR-0115): its terminal record and every per-instance family are removed from the state store through a durable IntentPurged event, so the deletion replays on recovery. The retention sweep calls it for an instance it has already read from the history index and gated on age + exported position; the carried value supplies the definition key the cleanup needs. Purging an instance that is not (or no longer) in history is a harmless no-op — the deletes are idempotent. Call RunUntilIdle to process it.

func (*Processor) Recover 

func (p *Processor) Recover() error

Recover rebuilds in-memory position/key state and catches the store up to the log. It replays events after the store's last applied position through the same applyToState used live (invariant I4), and restores the key counter and log position from what the log already froze (invariant I6). Call once after New, before processing.

func (*Processor) RecoverFrom added in v0.2.0 

func (p *Processor) RecoverFrom(checkpointRoot string) error

RecoverFrom is Recover with a recovery-checkpoint root (ADR-0131). When the root holds a checkpoint this processor may skip past, replay starts after the position that checkpoint covers instead of at genesis, and the highest log position and key counter it recorded seed what the skipped prefix would have contributed.

An empty root, or no usable checkpoint, replays the whole log exactly as before: falling back is always correct, only slower, so a missing, corrupt, foreign, or too-new checkpoint can never produce wrong state (invariant I2 is untouched — the WAL remains the source of truth).

func (*Processor) ResolveIncident 

func (p *Processor) ResolveIncident(elementKey uint64, retries int32)

ResolveIncident enqueues an operator's resolution of the incident attached to elementKey (ADR-0061): the incident is cleared and its job re-created with retries (>= 1), returning it to the activatable index so a worker retries it. Resolving an incident that no longer exists is a no-op. Call RunUntilIdle (or Drive) to process it.

func (*Processor) RunUntilIdle 

func (p *Processor) RunUntilIdle() error

RunUntilIdle processes batches until the queue (including generated followups) drains. Deterministic and synchronous — the basis for tests and simple embedding; the channel-driven concurrent loop arrives with the API milestone.

func (*Processor) SetCallTargetOverride 

func (p *Processor) SetCallTargetOverride(calledProcessId string, ov CallTargetOverride)

SetCallTargetOverride installs (or replaces) the per-server override for a called process id. Must be called on the run-loop goroutine (the map's single owner): the server layer calls it at startup and on an admin change (ADR-0105).

func (*Processor) SetExecutionBudget added in v0.5.0 

func (p *Processor) SetExecutionBudget(steps int32)

SetExecutionBudget sets how many steps one token may take in a single run. A value of zero or less restores DefaultExecutionBudget; there is deliberately no way to turn the budget off, because "off" is the behaviour this exists to remove.

func (*Processor) SetJobNotifier 

func (p *Processor) SetJobNotifier(fn func(jobType int32))

SetJobNotifier installs the hook the service-task behavior triggers (after fsync) when a job of a type becomes available.

func (*Processor) SetMaxCollection added in v0.6.0 

func (p *Processor) SetMaxCollection(n int64)

SetMaxCollection sets how large an output collection may be. Zero or less restores DefaultMaxCollection.

func (*Processor) SetMaxIterations added in v0.5.0 

func (p *Processor) SetMaxIterations(n int)

SetMaxIterations sets how many iterations one multi-instance activity may ask for. Zero or less restores DefaultMaxIterations.

func (*Processor) SetMaxVariable added in v0.6.0 

func (p *Processor) SetMaxVariable(n int64)

SetMaxVariable sets how large one variable's value may be. Zero or less restores DefaultMaxVariable; as with every budget here there is no way to turn it off.

func (*Processor) SetMetrics added in v0.2.0 

func (p *Processor) SetMetrics(m Metrics)

SetMetrics attaches batch instrumentation, or detaches it with nil. Call it before the processor starts handling commands; like SetJobNotifier it is not safe to change while batches are running.

func (*Processor) SetProcessActive added in v0.2.0 

func (p *Processor) SetProcessActive(defKey uint64, active bool)

SetProcessActive marks a deployed definition active or inactive (ADR-0119). An inactive definition does not auto-start new instances when its timer, message, or signal start events fire; existing instances run to completion, and an explicit operator/API start is unaffected. The server layer loads the flag from the deployment sidecar at startup and calls this on an operator toggle. Like SetCallTargetOverride it is operator config, not event-sourced: it changes only the live decision to schedule a create, so replay is unaffected (I6). Run-loop goroutine only (the map's single owner).

func (*Processor) SetVariables 

func (p *Processor) SetVariables(piKey, scopeKey uint64, actor string, vars ...model.VariableValue)

SetVariables enqueues an external, operator-initiated write of variables onto a running instance's scope (ADR-0095): each variable is created if its name is new in the target scope or overwritten if it already exists. piKey is the process instance; scopeKey is the scope the variables land in — pass piKey (or 0, which the handler treats as piKey) for the instance root scope, or a live element instance key belonging to piKey for a subprocess/multi-instance-body local scope. The writes are frozen into VariableCreated/VariableUpdated events, so they replay without re-running this command (invariant I6) and appear in the instance's variable timeline as the audit trail. Setting variables on an instance that is gone (finished or never existed), or on a scope that does not belong to it, is a no-op. It does not re-evaluate any gateway a token has already passed — it only changes the stored values. Each variable set is additionally recorded as an audit event naming actor — who made the change (ADR-0098) — so the "who changed it" trail is durable; pass "" when the caller is unidentified. Call RunUntilIdle (or Drive) to process it.

func (*Processor) ThrowJobError 

func (p *Processor) ThrowJobError(jobKey uint64, errorCode string)

ThrowJobError enqueues a worker's report that its job threw a BPMN error code (ADR-0089) — the "throw BPMN error" verb, a sibling of FailJob. Instead of retrying or raising an incident, the handler cancels the job and propagates the error from the job's element to the nearest matching error boundary or error event subprocess (or, uncaught, raises an incident). The code rides in the command's incident.Message field, a transient command carrier that is never persisted. Throwing on a job that no longer exists is a no-op. Call RunUntilIdle (or Drive) to process it.

func (*Processor) TickTimers 

func (p *Processor) TickTimers() error

TickTimers fires all due timers and processes the resulting work to idle. A server scheduler calls it on the partition's goroutine (invariant I3).

func (*Processor) TriggerDueTimers 

func (p *Processor) TriggerDueTimers() error

TriggerDueTimers enqueues a trigger command for every timer due at or before the current clock, carrying each timer's value so the handler needs no extra read. Call RunUntilIdle (or TickTimers) to process them. It is time-driven, so it belongs off the command path — a scheduler calls it periodically.

func (*Processor) Undeploy 

func (p *Processor) Undeploy(defKey uint64)

Undeploy removes a definition so no new instances of it can be created, dropping its message-start index entries too. It is the caller's responsibility not to undeploy a definition with running instances (they resolve their definition by key on every batch).

type RecoveryStats added in v0.2.0 

type RecoveryStats struct {
	Seconds  float64
	Replayed int
	// Done is false on a processor that has not recovered yet, so a reader can tell
	// "no recovery" from "a recovery that read nothing".
	Done bool
}

RecoveryStats is what the last recovery did: how long it took and how many records it read from the log. It answers the question a restart raises — "how long was this down, and why?" — and, alongside the checkpoint gauges (ADR-0131), whether the checkpoint cadence is actually shortening replay.

Replayed counts records *read*, not events applied: a record at or below the store's applied position is skipped rather than folded in, and a checkpoint lets recovery skip whole segments without reading them at all. That is the number the cadence changes.

type SystemClock 

type SystemClock struct{}

SystemClock reads the host clock.

func (SystemClock) Now 

func (SystemClock) Now() int64

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.