Documentation
¶
Overview ¶
Package run provides persistence types for workflow runs.
Use Store to persist RunRecord values. MemoryStore is intended for tests, demos, and local examples (it is not durable).
For durable backends, use github.com/justinush/maestro/pkg/run/postgres.Store (Postgres + JSONB). Ensure workflow_runs exists before use: optional [postgres.ApplySchema], or your own migrations ([postgres.SchemaDDL] / schema.sql).
Build records from a live instance with RecordFromInstance, then restore with github.com/justinush/maestro/pkg/maestro.Runtime.RestoreInstance (preferred) or InstanceFromRecord for custom Store integrations.
Store.Save uses optimistic locking via RunRecord.Revision; stale writes return ErrRevisionConflict.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrNotFound is returned when no run exists for the given id. ErrNotFound = errors.New("run: not found") // ErrExists is returned when Create is called for an id that already exists. ErrExists = errors.New("run: already exists") // ErrRevisionConflict is returned when Save is called with a stale revision. ErrRevisionConflict = errors.New("run: revision conflict") )
Errors returned by Store. Persistence uses optimistic locking on RunRecord.Revision.
Functions ¶
func InstanceFromRecord ¶
func InstanceFromRecord(rec *RunRecord, def *definition.WorkflowDefinition, opts engine.Options) (*engine.Instance, error)
InstanceFromRecord restores an engine.Instance from a RunRecord.
Lower-level API: application code should prefer maestro.Runtime.RestoreInstance, which binds the record to the validated Runtime workflow definition.
Types ¶
type MemoryStore ¶
type MemoryStore struct {
// contains filtered or unexported fields
}
MemoryStore is an in-process Store (good for tests and demos; not durable).
func NewMemoryStore ¶
func NewMemoryStore() *MemoryStore
NewMemoryStore creates an empty memory-backed store.
func (*MemoryStore) Create ¶
func (m *MemoryStore) Create(ctx context.Context, rec *RunRecord) error
Create inserts rec when RunID is unused; stored revision becomes 1.
type RunRecord ¶
type RunRecord struct {
// RunID is the primary key for Store implementations.
RunID string `json:"runId"`
// WorkflowID matches WorkflowDefinition.ID.
WorkflowID string `json:"workflowId"`
// WorkflowVersion matches WorkflowDefinition.Version.
WorkflowVersion string `json:"workflowVersion"`
// Revision is the optimistic-lock token; see Store.Save.
Revision int64 `json:"revision"`
// State is the engine snapshot restored by InstanceFromRecord or RestoreInstance.
State engine.Snapshot `json:"state"`
}
RunRecord is the persistence shape for a workflow run.
JSON field names (runId, workflowId, revision, state, ...) are stable for v0.x; treat them as part of the public persistence contract.
Revision supports optimistic locking on Store.Save:
- Store.Create stores the record with revision 1 (ignore rec.Revision on input).
- Store.Get returns the current revision.
- Store.Save succeeds only when rec.Revision matches the stored value, then increments it.
- run.ErrRevisionConflict when another writer updated the run first.
func RecordFromInstance ¶
func RecordFromInstance(in *engine.Instance, def *definition.WorkflowDefinition, revision int64) *RunRecord
RecordFromInstance builds a RunRecord from a live instance.
Pass revision 0 before Store.Create (Create sets revision to 1). Before Store.Save, set revision to the value returned by Store.Get.
Example ¶
package main
import (
"context"
"fmt"
"log"
"github.com/justinush/maestro/pkg/definition"
"github.com/justinush/maestro/pkg/engine"
"github.com/justinush/maestro/pkg/run"
)
func main() {
def := &definition.WorkflowDefinition{
SchemaVersion: "0.1",
ID: "w",
Version: "1",
InitialStepID: "a",
TerminalStepIDs: []string{"end"},
Steps: []definition.Step{
{ID: "a", Kind: definition.StepKindAction},
{ID: "end", Kind: definition.StepKindEnd},
},
Transitions: []definition.Transition{{From: "a", To: "end"}},
}
in, err := engine.NewInstance(def, engine.Options{RunID: "run-1"})
if err != nil {
log.Fatal(err)
}
if res := in.RunUntilBlocked(); res.Status != engine.RunCompleted {
log.Fatalf("status: %v err=%v", res.Status, res.Err)
}
st := run.NewMemoryStore()
rec := run.RecordFromInstance(in, def, 0)
rec.RunID = "run-1"
if err := st.Create(context.Background(), rec); err != nil {
log.Fatal(err)
}
got, err := st.Get(context.Background(), "run-1")
if err != nil {
log.Fatal(err)
}
fmt.Println(got.Revision, got.State.CurrentStepID)
}
Output: 1 end
type Store ¶
type Store interface {
// Create inserts a new run. rec.RunID must be set; rec.Revision is stored as 1.
Create(ctx context.Context, rec *RunRecord) error
// Get returns a deep copy of the run or ErrNotFound.
Get(ctx context.Context, runID string) (*RunRecord, error)
// Save updates the run when rec.Revision matches the stored value (then bumps revision).
Save(ctx context.Context, rec *RunRecord) error
}
Store persists RunRecord values. Implementations must be safe for concurrent use.
Revision rules (see RunRecord):
- Create: inserts a new run; stored revision becomes 1.
- Get: returns a copy including the current revision.
- Save: updates only when rec.Revision matches; on success revision increments by 1.
Source Files
Directories