engine

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 11, 2026 License: AGPL-3.0 Imports: 10 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

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

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
	// 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
	// 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
	// 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
}

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 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)

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 API 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.

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) 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)

AppendVariableEvent records a variable write. 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) ElementInstancesOnNode 

func (c *ProcessingContext) ElementInstancesOnNode(procKey uint64, elementId int32) []uint64

ElementInstancesOnNode returns the keys of every live element instance sitting on the given BPMN node within a process instance, seen through the in-flight transaction (so it includes one activated earlier in this batch). A parallel join uses it to count how many tokens have arrived on its incoming flows.

func (*ProcessingContext) ForEachActiveProcessInstance 

func (c *ProcessingContext) ForEachActiveProcessInstance(fn func(piKey uint64, pi *model.ProcessInstanceValue))

ForEachActiveProcessInstance calls fn with the key and value of every live process instance, via the committed process-instance column family. Entries are collected before fn runs so fn may emit events/commands (e.g. terminate a child) without disturbing the scan. Used to find the child a call activity started, via its persisted parent link (ADR-0076).

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, via the committed elByProc index. 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) 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 uint64, nodeId int32, reaches map[int32]bool) bool

TokenCanStillReach reports whether any live token 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.

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) 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) 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) 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) 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) 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)

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).

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) 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) 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) 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) 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) 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 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.