Help us shape the future of Go! Take the 2026 Go Developer Survey

storage

package
v0.7.0 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package storage provides the storage layer for Selvedge.

Package storage provides the storage layer for Selvedge.

Package storage provides the storage layer for Selvedge.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidSignalChannel = errors.New("signal channel must contain 1 to 255 bytes")
	ErrSignalNotAvailable   = errors.New("signal is absent, consumed, or belongs to another instance or channel")
)
View Source 
var (
	ErrWorkflowAlreadyCompleted = errors.New("workflow already completed")
	ErrWorkflowAlreadyFailed    = errors.New("workflow already failed")
	ErrWorkflowAlreadyCancelled = errors.New("workflow already cancelled")
)

Status errors retain identity across the storage and public APIs.

View Source 
var ErrDeliveryConflict = errors.New("delivery candidate is absent or already delivered")
View Source 
var ErrFenced = errors.New("execution fenced out")

ErrFenced means a write lacks the current instance lease token.

View Source 
var ErrHistoryKeyConflict = errors.New("history key holds another command")
View Source 
var ErrInstanceClosed = errors.New("workflow instance already closed")

ErrInstanceClosed indicates that a close cannot change a terminal instance.

View Source 
var ErrInstanceNotFound = errors.New("workflow instance not found")

ErrInstanceNotFound indicates that an instance ID does not exist.

View Source 
var ErrWorkflowAlreadyStarted = errors.New("workflow already started")

Functions

func ContextWithCommitCheck 

func ContextWithCommitCheck(ctx context.Context, check func() error) context.Context

ContextWithCommitCheck checks the holder's monotonic belief before guarded commits.

func ContextWithFence 

func ContextWithFence(ctx context.Context, f Fence) context.Context

ContextWithFence carries the execution token for status and subscription writes.

func ContextWithFenceRejection 

func ContextWithFenceRejection(ctx context.Context, reject func(error)) context.Context

ContextWithFenceRejection makes rejected commits fatal to their execution.

func InitializeTestSchema 

func InitializeTestSchema(ctx context.Context, s *SQLiteStorage) error

InitializeTestSchema applies the embedded SQLite schema used by the App. This is intended for use in tests only.

func InitializeTestSchemaForStorage 

func InitializeTestSchemaForStorage(ctx context.Context, s Storage) error

InitializeTestSchemaForStorage initializes the embedded schema for a Storage. This is useful for external test packages.

func InitializeTestSchemaPostgres 

func InitializeTestSchemaPostgres(ctx context.Context, s *PostgresStorage) error

InitializeTestSchemaPostgres applies the embedded PostgreSQL schema used by the App. This is intended for use in tests only.

func SubscribeLocal 

func SubscribeLocal(database string, receive func(LocalNotification)) func()

SubscribeLocal subscribes to one database's process-local notifications. Callbacks must not block. The returned unsubscribe function is idempotent. Memory databases are excluded because their identity is connection-specific.

func ValidateJSONPath 

func ValidateJSONPath(path string) error

ValidateJSONPath validates a JSON path for safe use in SQL queries. Returns an error if the path contains potentially dangerous characters.

func WithTestStatementObserver 

func WithTestStatementObserver(observe func(string, []any)) func(Storage)

WithTestStatementObserver records the SQL protocol without replacing the store. Install before concurrent access. The observer must be concurrency-safe.

func WithTestStoreClock 

func WithTestStoreClock() func(Storage)

WithTestStoreClock selects a clock table created by the test harness. Apply only before concurrent access. Production schemas never create the table.

Types

type ActivityTask 

type ActivityTask struct {
	ID                                                    int64
	InstanceID, ActivityID                                string
	Seq                                                   int64
	ActivityName                                          string
	InputData                                             []byte
	Queue                                                 string
	Options                                               []byte
	MaxAttempts, Failures                                 int
	NotBeforeMS                                           int64
	LastError                                             []byte
	Status                                                string
	Attempts                                              int
	ClaimOwner                                            string
	ClaimEpoch                                            int64
	ClaimExpiresMS                                        *int64
	ScheduleToStartMS, StartToCloseMS, HeartbeatTimeoutMS int64
	ScheduleDeadlineMS, AttemptDeadlineMS                 *int64
	ClaimStarted                                          time.Time
}

ActivityTask is the durable input and current claim for one activity.

func (*ActivityTask) Claim 

func (t *ActivityTask) Claim() TaskClaim

Claim returns the attempt token stored on this task.

type ActivityTaskManager 

type ActivityTaskManager interface {
	CancelActivityAtYield(context.Context, *Fence, string, string, string, []byte) (*HistoryEvent, error)
	ReserveActivity(context.Context, *Fence, int64, string, string, *ActivityTask) (*HistoryEvent, error)
	// ClaimActivityTasks requires registered names; a nil or empty filter claims nothing.
	ClaimActivityTasks(context.Context, string, time.Duration, int, TaskFilter) ([]*ActivityTask, error)
	ClaimActivityTask(context.Context, string, string, string, time.Duration, []string) (*ActivityTask, error)
	RenewActivityTask(context.Context, TaskClaim, time.Duration, bool) error
	CompleteActivityTask(context.Context, TaskClaim, HistoryEventType, []byte) error
	ReleaseActivityTask(context.Context, TaskClaim) error
	RetryActivityTask(context.Context, TaskClaim, []byte, time.Duration) error
	NextActivityDue(context.Context, TaskFilter) (time.Duration, bool, error)
	ListActivityTasks(context.Context, string) ([]*ActivityTask, error)
	FindActivityTimeouts(context.Context, int) ([]*ActivityTimeout, error)
	TimeoutActivityTask(context.Context, *ActivityTimeout, []byte) error
}

ActivityTaskManager persists and fences durable activity attempts.

type ActivityTimeout 

type ActivityTimeout struct {
	Task      *ActivityTask
	Kind      string
	Exhausted bool
}

ActivityTimeout is a snapshot candidate for a fenced timeout write.

type ChildCancel 

type ChildCancel struct {
	Seq                             int64
	Record                          *HistoryEvent
	ChildKey, ChildCommand, ChildID string
	EventData                       []byte
}

ChildCancel records an explicit cancel at Seq, or a Get observation at zero. Record is the explicit child_cancel command; EventData is the cancellation envelope.

type ChildManager 

type ChildManager interface {
	StartChild(context.Context, *Fence, int64, string, string, *WorkflowInstance) (*HistoryEvent, error)
	CancelChild(context.Context, *Fence, *ChildCancel) (*HistoryEvent, error)
}

ChildManager atomically connects child instances to their parent's history.

type ChildReservation 

type ChildReservation struct {
	ChildID  string `json:"child_id"`
	Workflow string `json:"workflow"`
}

ChildReservation identifies the immutable child belonging to a command.

type DeliveryBatch 

type DeliveryBatch struct {
	HistoryKeys []string
	Signals     []SignalRef
}

type DeliveryManager 

type DeliveryManager interface {
	ArrivedUndelivered(context.Context, string) (*DeliveryBatch, error)
	DeliveredSignals(context.Context, string) ([]SignalPosition, error)
	CommitObservation(context.Context, *Fence, *ObservationCommit) (*ObservationResult, error)
}

type Driver 

type Driver interface {
	NowMillisExpr() string
	// DriverName returns the driver name (e.g., "sqlite", "postgres")
	DriverName() string

	// GetCurrentTimeExpr returns the SQL expression for current time.
	// SQLite: datetime('now'), PostgreSQL: NOW()
	GetCurrentTimeExpr() string

	// MakeDatetimeComparable wraps a datetime column for comparison.
	// SQLite needs datetime() wrapper, PostgreSQL doesn't.
	MakeDatetimeComparable(column string) string

	// SelectForUpdateSkipLocked returns the clause for row locking.
	// PostgreSQL: "FOR UPDATE SKIP LOCKED", SQLite: "" (uses table-level locking)
	SelectForUpdateSkipLocked() string

	// Placeholder returns the placeholder for the nth parameter.
	// SQLite: ?, PostgreSQL: $n
	Placeholder(n int) string

	// OnConflictDoNothing returns the clause for idempotent inserts.
	// SQLite: "ON CONFLICT DO NOTHING", PostgreSQL: "ON CONFLICT DO NOTHING"
	OnConflictDoNothing(conflictColumns ...string) string

	// ReturningClause returns the RETURNING clause for insert/update.
	// SQLite: "", PostgreSQL: "RETURNING id"
	ReturningClause(columns ...string) string

	// JSONExtract returns the SQL expression to extract a value from a JSON column.
	// The path is a dot-separated path like "order.customer.id".
	// The returned expression extracts the value as text.
	JSONExtract(column, path string) string

	// JSONCompare returns the SQL expression to compare a JSON-extracted value
	// with a given value, handling type coercion appropriately.
	// Returns the comparison expression and any additional args needed.
	JSONCompare(extractExpr string, value any, placeholderNum int) (expr string, args []any)
}

Driver abstracts database-specific SQL operations.

func NewDriver 

func NewDriver(dbURL string) Driver

NewDriver creates a new driver based on the database URL.

type Executor 

type Executor interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

Executor is a database executor interface that can be either *sql.DB or *sql.Tx. This allows users to execute custom SQL queries within the same transaction as the workflow activity.

type Fence 

type Fence struct {
	InstanceID, Owner string
	Epoch             int64
}

Fence identifies one execution lease.

type HistoryEvent 

type HistoryEvent struct {
	DeliveredSeq    int64            `json:"delivered_seq"`
	Seq             int64            `json:"seq"`                        // Zero represents SQL NULL for lifecycle rows.
	ReservationData []byte           `json:"reservation_data,omitempty"` // Immutable command metadata, separate from its observed value.
	Command         string           `json:"command"`
	ID              int64            `json:"id"`
	InstanceID      string           `json:"instance_id"`
	ActivityID      string           `json:"activity_id"`
	EventType       HistoryEventType `json:"event_type"`
	EventData       []byte           `json:"event_data"`        // JSON-encoded data
	EventDataBinary []byte           `json:"event_data_binary"` // Binary data (alternative)
	DataType        string           `json:"data_type"`         // "json" or "binary"
	CreatedAt       time.Time        `json:"created_at"`
}

HistoryEvent represents a single event in a workflow's execution history.

type HistoryEventType 

type HistoryEventType string

HistoryEventType represents the type of history event. 

const (
	HistoryChildCompleted   HistoryEventType = "ChildCompleted"
	HistoryChildFailed      HistoryEventType = "ChildFailed"
	HistoryChildCancelled   HistoryEventType = "ChildCancelled"
	HistoryChildStartFailed HistoryEventType = "ChildStartFailed"
)
const (
	HistoryScopeCancelled HistoryEventType = "ScopeCancelled"
	HistoryReserved       HistoryEventType = "Reserved"

	// Activity events
	HistoryActivityCancelled HistoryEventType = "ActivityCancelled"
	HistoryActivityCompleted HistoryEventType = "ActivityCompleted"
	HistoryActivityFailed    HistoryEventType = "ActivityFailed"

	// Timer expired (via Sleep/SleepUntil)
	HistoryTimerExpired    HistoryEventType = "TimerExpired"
	HistoryTimerCancelled  HistoryEventType = "TimerCancelled"
	HistoryTimerSuperseded HistoryEventType = "TimerSuperseded"

	// Signal observations contain the payload or a cancellation envelope.
	HistorySignalReceived  HistoryEventType = "SignalReceived"
	HistorySignalCancelled HistoryEventType = "SignalCancelled"

	HistoryAwaitCompleted HistoryEventType = "AwaitCompleted"
	HistoryAwaitCancelled HistoryEventType = "AwaitCancelled"
	HistorySignalNone     HistoryEventType = "SignalNone"
	HistorySignalTimeout  HistoryEventType = "SignalTimeout"

	// Workflow lifecycle events
	HistoryWorkflowFailed    HistoryEventType = "WorkflowFailed"
	HistoryWorkflowCancelled HistoryEventType = "WorkflowCancelled"
)
const HistoryNowRecorded HistoryEventType = "NowRecorded"

HistoryNowRecorded is a recorded workflow wall-clock observation.

type HistoryFill 

type HistoryFill struct {
	Key, Command string
	EventType    HistoryEventType
	EventData    []byte
}

type HistoryIterator 

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

HistoryIterator provides streaming access to workflow history events. This avoids loading the entire history into memory at once.

func NewHistoryIterator 

func NewHistoryIterator(ctx context.Context, storage Storage, instanceID string, opts *HistoryIteratorOptions) *HistoryIterator

NewHistoryIterator creates a new history iterator.

func (*HistoryIterator) Close 

func (it *HistoryIterator) Close() error

Close releases any resources held by the iterator. Currently a no-op, but provided for future compatibility.

func (*HistoryIterator) Collect 

func (it *HistoryIterator) Collect() ([]*HistoryEvent, error)

Collect reads all remaining events into a slice. Use with caution for large histories - prefer iterating with Next().

func (*HistoryIterator) Err 

func (it *HistoryIterator) Err() error

Err returns any error that occurred during iteration.

func (*HistoryIterator) Next 

func (it *HistoryIterator) Next() (*HistoryEvent, bool)

Next returns the next history event. Returns (event, true) if an event is available, (nil, false) if no more events. Check Err() after iteration to see if an error occurred.

type HistoryIteratorOptions 

type HistoryIteratorOptions struct {
	// BatchSize is the number of events to fetch per batch.
	// Default: 1000
	BatchSize int
}

HistoryIteratorOptions configures the history iterator.

type HistoryManager 

type HistoryManager interface {
	// ReserveHistory inserts a command with a SQL NULL value; matching repeats return the stored row.
	ReserveHistory(ctx context.Context, f *Fence, instanceID, key, cmd string, seq int64) (*HistoryEvent, error)
	// RecordHistory inserts a command and value. Matching repeats preserve the entire first row.
	RecordHistory(ctx context.Context, f *Fence, event *HistoryEvent) (*HistoryEvent, error)
	// SetHistoryValueIfNull only fills an existing reservation; the first value wins.
	SetHistoryValueIfNull(ctx context.Context, f *Fence, instanceID, key, cmd string, t HistoryEventType, data []byte) (*HistoryEvent, error)
	// GetHistoryEvent returns nil, nil when the key does not exist.
	GetHistoryEvent(ctx context.Context, instanceID, key string) (*HistoryEvent, error)

	// GetHistoryPaginated retrieves history events for an instance with pagination.
	// Returns events with id > afterID, up to limit events.
	// Returns (events, hasMore, error).
	GetHistoryPaginated(ctx context.Context, instanceID string, afterID int64, limit int) ([]*HistoryEvent, bool, error)

	// GetHistoryCount returns the total number of history events for an instance.
	GetHistoryCount(ctx context.Context, instanceID string) (int64, error)
}

HistoryManager handles workflow execution history.

type InputFilterBuilder 

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

InputFilterBuilder builds SQL query conditions for input filters.

func NewInputFilterBuilder 

func NewInputFilterBuilder(driver Driver) *InputFilterBuilder

NewInputFilterBuilder creates a new InputFilterBuilder.

func (*InputFilterBuilder) BuildFilterQuery 

func (b *InputFilterBuilder) BuildFilterQuery(inputFilters map[string]any, startPlaceholder int) (conditions []string, args []any, err error)

BuildFilterQuery builds the WHERE clause conditions for input filters. Returns the query parts (as a slice of conditions) and args to append, or an error if validation fails. The startPlaceholder is the starting placeholder number (for PostgreSQL).

type InstanceClosedError 

type InstanceClosedError struct {
	InstanceID string
	Status     WorkflowStatus
}

InstanceClosedError reports the status that prevented signal ingress.

func (*InstanceClosedError) Error 

func (e *InstanceClosedError) Error() string

func (*InstanceClosedError) Is 

func (e *InstanceClosedError) Is(target error) bool

Is also matches the sentinel for this terminal status.

func (*InstanceClosedError) Unwrap 

func (e *InstanceClosedError) Unwrap() error

type InstanceManager 

type InstanceManager interface {
	// CreateInstance creates a new workflow instance.
	CreateInstance(ctx context.Context, instance *WorkflowInstance) error

	// GetInstance retrieves a workflow instance by ID.
	GetInstance(ctx context.Context, instanceID string) (*WorkflowInstance, error)

	// RequestCancel durably requests cancellation and wakes a running instance.
	RequestCancel(ctx context.Context, instanceID, reason string) (WorkflowStatus, error)

	// IsCancelRequested reports whether cancellation was requested.
	IsCancelRequested(ctx context.Context, instanceID string) (bool, error)

	// CloseInstance atomically closes a running instance under its execution fence.
	CloseInstance(ctx context.Context, f Fence, status WorkflowStatus, output []byte, closeEvent *HistoryEvent) error

	// UpdateInstanceActivity updates the current activity ID.
	UpdateInstanceActivity(ctx context.Context, instanceID string, activityID string) error

	// ListInstances lists workflow instances with optional filtering and pagination.
	// Returns PaginationResult with cursor-based pagination.
	ListInstances(ctx context.Context, opts ListInstancesOptions) (*PaginationResult, error)
}

InstanceManager handles workflow instance CRUD operations.

type InstanceSnapshot 

type InstanceSnapshot struct {
	Instance        *WorkflowInstance
	CancelRequested bool
	History         []*HistoryEvent
	Signals         []SignalPosition
}

InstanceSnapshot is one consistent, detached view of an instance and its replay inputs.

type LeaseManager 

type LeaseManager interface {
	// AcquireLease grants a fresh epoch to a runnable unowned instance or an instance with an expired lease.
	AcquireLease(context.Context, string, string, time.Duration) (Fence, bool, error)
	// RenewLease extends an unexpired matching lease using the store clock.
	RenewLease(context.Context, Fence, time.Duration) (bool, error)
	// ReleaseLease clears a matching lease and preserves pending input; runnable also requests another dispatch.
	ReleaseLease(context.Context, Fence, bool) error
	// MarkRunnable marks a running instance for dispatch without changing ownership.
	MarkRunnable(context.Context, string) error
	// FindDispatchable lists runnable unowned instances and instances with expired leases.
	// A nil name filter includes all workflows; a non-nil empty filter includes none.
	FindDispatchable(context.Context, int, ...string) ([]string, error)
}

LeaseManager grants a distinct token to each execution using the store clock.

type ListInstancesOptions 

type ListInstancesOptions struct {
	// Pagination
	Limit     int    // Page size (default: 50)
	PageToken string // Cursor token for pagination (format: "ISO_DATETIME||INSTANCE_ID")

	// Filters
	StatusFilter       WorkflowStatus // Filter by status
	WorkflowNameFilter string         // Filter by workflow name (partial match, case-insensitive)
	InstanceIDFilter   string         // Filter by instance ID (partial match, case-insensitive)
	StartedAfter       *time.Time     // Filter instances started after this time
	StartedBefore      *time.Time     // Filter instances started before this time
	InputFilters       map[string]any // Filter by JSON paths in input data (exact match)

	// Deprecated: Use StatusFilter instead
	Status WorkflowStatus
	// Deprecated: Use WorkflowNameFilter instead
	WorkflowName string
	// Deprecated: Use PageToken instead
	Offset int
}

ListInstancesOptions defines options for listing instances.

type LocalNotification 

type LocalNotification struct {
	InstanceID string
	Closed     bool
}

LocalNotification is a hint to reread durable instance state.

type ObservationCommit 

type ObservationCommit struct {
	Cancel     *ScopeCancel
	Seq        int64
	Batch      *DeliveryBatch
	DeliverAll bool
	Record     *HistoryEvent
	Fill       *HistoryFill
	Consume    *SignalConsume
	Retire     *TimerRetire
}

type ObservationResult 

type ObservationResult struct {
	Stored    *HistoryEvent
	Delivered DeliveryBatch
}

type PaginationResult 

type PaginationResult struct {
	Instances     []*WorkflowInstance `json:"instances"`
	NextPageToken string              `json:"next_page_token,omitempty"`
	HasMore       bool                `json:"has_more"`
}

PaginationResult represents a paginated list of workflow instances.

type PostgresDriver 

type PostgresDriver struct{}

PostgresDriver implements Driver for PostgreSQL.

func (*PostgresDriver) DriverName 

func (d *PostgresDriver) DriverName() string

func (*PostgresDriver) GetCurrentTimeExpr 

func (d *PostgresDriver) GetCurrentTimeExpr() string

func (*PostgresDriver) JSONCompare 

func (d *PostgresDriver) JSONCompare(extractExpr string, value any, placeholderNum int) (expr string, compArgs []any)

JSONCompare for PostgresDriver returns the SQL expression to compare a JSON-extracted value.

func (*PostgresDriver) JSONExtract 

func (d *PostgresDriver) JSONExtract(column, path string) string

JSONExtract for PostgresDriver returns the SQL expression to extract a value from a JSON column.

func (*PostgresDriver) MakeDatetimeComparable 

func (d *PostgresDriver) MakeDatetimeComparable(column string) string

func (*PostgresDriver) NowMillisExpr 

func (d *PostgresDriver) NowMillisExpr() string

NowMillisExpr returns SQL for the store clock in Unix milliseconds.

func (*PostgresDriver) OnConflictDoNothing 

func (d *PostgresDriver) OnConflictDoNothing(conflictColumns ...string) string

func (*PostgresDriver) Placeholder 

func (d *PostgresDriver) Placeholder(n int) string

func (*PostgresDriver) ReturningClause 

func (d *PostgresDriver) ReturningClause(columns ...string) string

func (*PostgresDriver) SelectForUpdateSkipLocked 

func (d *PostgresDriver) SelectForUpdateSkipLocked() string

type PostgresStorage 

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

PostgresStorage implements the Storage interface using PostgreSQL.

func NewPostgresStorage 

func NewPostgresStorage(connStr string) (*PostgresStorage, error)

NewPostgresStorage creates a new PostgreSQL storage. The connStr should be a PostgreSQL connection string: "postgres://user:password@localhost:5432/dbname?sslmode=disable"

func (*PostgresStorage) AcquireLease 

func (s *PostgresStorage) AcquireLease(ctx context.Context, id, owner string, ttl time.Duration) (Fence, bool, error)

AcquireLease grants a fresh epoch to a runnable unowned instance or an instance with an expired lease.

func (*PostgresStorage) AppendSignal 

func (s *PostgresStorage) AppendSignal(ctx context.Context, instanceID, channel string, payload []byte) (int64, error)

AppendSignal buffers a signal only while the instance is running.

func (*PostgresStorage) ArrivedUndelivered 

func (s *PostgresStorage) ArrivedUndelivered(ctx context.Context, id string) (*DeliveryBatch, error)

func (*PostgresStorage) BeginTransaction 

func (s *PostgresStorage) BeginTransaction(ctx context.Context) (context.Context, error)

BeginTransaction starts a new transaction.

func (*PostgresStorage) CancelActivityAtYield 

func (s *PostgresStorage) CancelActivityAtYield(ctx context.Context, f *Fence, id, key, cmd string, data []byte) (*HistoryEvent, error)

CancelActivityAtYield records cancellation and cancels the associated task atomically.

func (*PostgresStorage) CancelChild 

func (s *PostgresStorage) CancelChild(ctx context.Context, f *Fence, c *ChildCancel) (*HistoryEvent, error)

CancelChild resolves a handle and requests cancellation atomically.

func (*PostgresStorage) ClaimActivityTask 

func (s *PostgresStorage) ClaimActivityTask(ctx context.Context, id, key, owner string, ttl time.Duration, queues []string) (*ActivityTask, error)

ClaimActivityTask delegates to the durable task store.

func (*PostgresStorage) ClaimActivityTasks 

func (s *PostgresStorage) ClaimActivityTasks(ctx context.Context, owner string, ttl time.Duration, limit int, filter TaskFilter) ([]*ActivityTask, error)

ClaimActivityTasks delegates to the durable task store.

func (*PostgresStorage) CleanupExpiredSystemLocks 

func (s *PostgresStorage) CleanupExpiredSystemLocks(ctx context.Context) error

CleanupExpiredSystemLocks removes expired system locks.

func (*PostgresStorage) Close 

func (s *PostgresStorage) Close() error

Close closes the database connection.

func (*PostgresStorage) CloseInstance 

func (s *PostgresStorage) CloseInstance(ctx context.Context, f Fence, status WorkflowStatus, output []byte, event *HistoryEvent) error

CloseInstance commits a terminal status, result, close event, and timer cleanup together.

func (*PostgresStorage) CommitObservation 

func (s *PostgresStorage) CommitObservation(ctx context.Context, f *Fence, c *ObservationCommit) (*ObservationResult, error)

func (*PostgresStorage) CommitTransaction 

func (s *PostgresStorage) CommitTransaction(ctx context.Context) error

CommitTransaction commits the current transaction.

func (*PostgresStorage) CompleteActivityTask 

func (s *PostgresStorage) CompleteActivityTask(ctx context.Context, claim TaskClaim, event HistoryEventType, data []byte) error

CompleteActivityTask delegates to the durable task store.

func (*PostgresStorage) Conn 

func (s *PostgresStorage) Conn(ctx context.Context) Executor

Conn returns the database executor for the current context.

func (*PostgresStorage) CreateInstance 

func (s *PostgresStorage) CreateInstance(ctx context.Context, instance *WorkflowInstance) error

CreateInstance creates a new workflow instance.

func (*PostgresStorage) DB 

func (s *PostgresStorage) DB() *sql.DB

DB returns the underlying database connection.

func (*PostgresStorage) DeliveredSignals 

func (s *PostgresStorage) DeliveredSignals(ctx context.Context, id string) ([]SignalPosition, error)

func (*PostgresStorage) FindActivityTimeouts 

func (s *PostgresStorage) FindActivityTimeouts(ctx context.Context, limit int) ([]*ActivityTimeout, error)

FindActivityTimeouts finds eligible timeout candidates.

func (*PostgresStorage) FindDispatchable 

func (s *PostgresStorage) FindDispatchable(ctx context.Context, limit int, workflowNames ...string) ([]string, error)

FindDispatchable lists runnable unowned instances and instances with expired leases.

func (*PostgresStorage) FindExpiredTimers 

func (s *PostgresStorage) FindExpiredTimers(ctx context.Context, limit int) ([]*TimerSubscription, error)

FindExpiredTimers finds expired timers.

func (*PostgresStorage) FireTimer 

func (s *PostgresStorage) FireTimer(ctx context.Context, timer *TimerSubscription) error

FireTimer atomically records a timer result and wake, or discards a terminal instance's timer.

func (*PostgresStorage) GetHistoryCount 

func (s *PostgresStorage) GetHistoryCount(ctx context.Context, instanceID string) (int64, error)

GetHistoryCount returns the total number of history events for an instance.

func (*PostgresStorage) GetHistoryEvent 

func (s *PostgresStorage) GetHistoryEvent(ctx context.Context, instanceID, key string) (*HistoryEvent, error)

GetHistoryEvent returns the stored row, or nil when absent.

func (*PostgresStorage) GetHistoryPaginated 

func (s *PostgresStorage) GetHistoryPaginated(ctx context.Context, instanceID string, afterID int64, limit int) ([]*HistoryEvent, bool, error)

GetHistoryPaginated retrieves history events with pagination. Returns events with id > afterID, up to limit events. Returns (events, hasMore, error).

func (*PostgresStorage) GetInstance 

func (s *PostgresStorage) GetInstance(ctx context.Context, instanceID string) (*WorkflowInstance, error)

GetInstance retrieves a workflow instance by ID.

func (*PostgresStorage) InTransaction 

func (s *PostgresStorage) InTransaction(ctx context.Context) bool

InTransaction returns whether a transaction is in progress.

func (*PostgresStorage) Initialize 

func (s *PostgresStorage) Initialize(ctx context.Context) error

Initialize is deprecated. Use dbmate for migrations: dbmate -d internal/migrations/sql/postgresql up

func (*PostgresStorage) IsCancelRequested 

func (s *PostgresStorage) IsCancelRequested(ctx context.Context, id string) (bool, error)

IsCancelRequested reads the durable cancellation request.

func (*PostgresStorage) IsNotifyEnabled 

func (s *PostgresStorage) IsNotifyEnabled() bool

IsNotifyEnabled returns whether PostgreSQL LISTEN/NOTIFY is enabled.

func (*PostgresStorage) ListActivityTasks 

func (s *PostgresStorage) ListActivityTasks(ctx context.Context, id string) ([]*ActivityTask, error)

ListActivityTasks returns tasks in reservation order.

func (*PostgresStorage) ListInstances 

func (s *PostgresStorage) ListInstances(ctx context.Context, opts ListInstancesOptions) (*PaginationResult, error)

ListInstances lists workflow instances with cursor-based pagination and filters.

func (*PostgresStorage) MarkRunnable 

func (s *PostgresStorage) MarkRunnable(ctx context.Context, id string) error

MarkRunnable marks a running instance for dispatch without changing ownership.

func (*PostgresStorage) NextActivityDue 

func (s *PostgresStorage) NextActivityDue(ctx context.Context, f TaskFilter) (time.Duration, bool, error)

NextActivityDue returns the delay until the next eligible queued task.

func (*PostgresStorage) NextSignal 

func (s *PostgresStorage) NextSignal(ctx context.Context, instanceID, channel string) (*Signal, error)

NextSignal returns the oldest unconsumed signal for this instance and channel.

func (*PostgresStorage) NotifyRunnable 

func (s *PostgresStorage) NotifyRunnable(ctx context.Context)

NotifyRunnable wakes dispatchers after a batch of runnable mutations, such as lease releases during shutdown. Local notifications remain enabled independently of PostgreSQL LISTEN/NOTIFY.

func (*PostgresStorage) ReadSnapshot 

func (s *PostgresStorage) ReadSnapshot(ctx context.Context, id string) (*InstanceSnapshot, error)

ReadSnapshot fixes all replay inputs to the same PostgreSQL MVCC snapshot.

func (*PostgresStorage) RecordHistory 

func (s *PostgresStorage) RecordHistory(ctx context.Context, f *Fence, ev *HistoryEvent) (*HistoryEvent, error)

RecordHistory returns the first stored command and value at the key.

func (*PostgresStorage) RegisterPostCommitCallback 

func (s *PostgresStorage) RegisterPostCommitCallback(ctx context.Context, cb func() error) error

RegisterPostCommitCallback registers a callback to be executed after a successful commit. Returns an error if not currently in a transaction.

func (*PostgresStorage) ReleaseActivityTask 

func (s *PostgresStorage) ReleaseActivityTask(ctx context.Context, claim TaskClaim) error

ReleaseActivityTask delegates to the durable task store.

func (*PostgresStorage) ReleaseLease 

func (s *PostgresStorage) ReleaseLease(ctx context.Context, f Fence, runnable bool) error

ReleaseLease clears a matching lease and preserves pending input; runnable also requests another dispatch.

func (*PostgresStorage) ReleaseSystemLock 

func (s *PostgresStorage) ReleaseSystemLock(ctx context.Context, lockName, workerID string) error

ReleaseSystemLock releases a system lock.

func (*PostgresStorage) RemoveTimerSubscription 

func (s *PostgresStorage) RemoveTimerSubscription(ctx context.Context, instanceID, timerID string) error

RemoveTimerSubscription removes a timer subscription.

func (*PostgresStorage) RenewActivityTask 

func (s *PostgresStorage) RenewActivityTask(ctx context.Context, claim TaskClaim, ttl time.Duration, extend bool) error

RenewActivityTask delegates to the durable task store.

func (*PostgresStorage) RenewLease 

func (s *PostgresStorage) RenewLease(ctx context.Context, f Fence, ttl time.Duration) (bool, error)

RenewLease extends an unexpired matching lease using the store clock.

func (*PostgresStorage) RequestCancel 

func (s *PostgresStorage) RequestCancel(ctx context.Context, id, reason string) (WorkflowStatus, error)

RequestCancel requests cancellation without acquiring an execution lease.

func (*PostgresStorage) ReserveActivity 

func (s *PostgresStorage) ReserveActivity(ctx context.Context, f *Fence, seq int64, key, cmd string, task *ActivityTask) (*HistoryEvent, error)

ReserveActivity delegates to the durable task store.

func (*PostgresStorage) ReserveHistory 

func (s *PostgresStorage) ReserveHistory(ctx context.Context, f *Fence, instanceID, key, cmd string, seq int64) (*HistoryEvent, error)

ReserveHistory reserves a command key with a NULL value.

func (*PostgresStorage) ReserveTimer 

func (s *PostgresStorage) ReserveTimer(ctx context.Context, f *Fence, seq int64, timer *TimerSubscription) (*HistoryEvent, error)

ReserveTimer reserves a sleep and arms its timer atomically under the execution fence.

func (*PostgresStorage) RetryActivityTask 

func (s *PostgresStorage) RetryActivityTask(ctx context.Context, c TaskClaim, data []byte, delay time.Duration) error

RetryActivityTask durably schedules another attempt.

func (*PostgresStorage) RollbackTransaction 

func (s *PostgresStorage) RollbackTransaction(ctx context.Context) error

RollbackTransaction rolls back the current transaction. Callbacks are not executed on rollback.

func (*PostgresStorage) SetHistoryValueIfNull 

func (s *PostgresStorage) SetHistoryValueIfNull(ctx context.Context, f *Fence, instanceID, key, cmd string, t HistoryEventType, data []byte) (*HistoryEvent, error)

SetHistoryValueIfNull records the first value for a matching command.

func (*PostgresStorage) SetNotifyEnabled 

func (s *PostgresStorage) SetNotifyEnabled(enabled bool)

SetNotifyEnabled enables or disables PostgreSQL LISTEN/NOTIFY. When enabled, the storage will send pg_notify() calls after key operations.

func (*PostgresStorage) StartChild 

func (s *PostgresStorage) StartChild(ctx context.Context, f *Fence, seq int64, key, cmd string, child *WorkflowInstance) (*HistoryEvent, error)

StartChild reserves the parent command and creates its child in one fenced transaction.

func (*PostgresStorage) TimeoutActivityTask 

func (s *PostgresStorage) TimeoutActivityTask(ctx context.Context, c *ActivityTimeout, data []byte) error

TimeoutActivityTask atomically records a final timeout.

func (*PostgresStorage) TryAcquireSystemLock 

func (s *PostgresStorage) TryAcquireSystemLock(ctx context.Context, lockName, workerID string, ttl time.Duration) (bool, error)

TryAcquireSystemLock attempts to acquire a system lock.

func (*PostgresStorage) UpdateInstanceActivity 

func (s *PostgresStorage) UpdateInstanceActivity(ctx context.Context, instanceID, activityID string) error

UpdateInstanceActivity updates the current activity ID.

type SQLiteDriver 

type SQLiteDriver struct{}

SQLiteDriver implements Driver for SQLite.

func (*SQLiteDriver) DriverName 

func (d *SQLiteDriver) DriverName() string

func (*SQLiteDriver) GetCurrentTimeExpr 

func (d *SQLiteDriver) GetCurrentTimeExpr() string

func (*SQLiteDriver) JSONCompare 

func (d *SQLiteDriver) JSONCompare(extractExpr string, value any, _ int) (expr string, compArgs []any)

JSONCompare for SQLiteDriver returns the SQL expression to compare a JSON-extracted value.

func (*SQLiteDriver) JSONExtract 

func (d *SQLiteDriver) JSONExtract(column, path string) string

JSONExtract for SQLiteDriver returns the SQL expression to extract a value from a JSON column.

func (*SQLiteDriver) MakeDatetimeComparable 

func (d *SQLiteDriver) MakeDatetimeComparable(column string) string

func (*SQLiteDriver) NowMillisExpr 

func (d *SQLiteDriver) NowMillisExpr() string

NowMillisExpr returns SQL for the store clock in Unix milliseconds.

func (*SQLiteDriver) OnConflictDoNothing 

func (d *SQLiteDriver) OnConflictDoNothing(conflictColumns ...string) string

func (*SQLiteDriver) Placeholder 

func (d *SQLiteDriver) Placeholder(n int) string

func (*SQLiteDriver) ReturningClause 

func (d *SQLiteDriver) ReturningClause(columns ...string) string

func (*SQLiteDriver) SelectForUpdateSkipLocked 

func (d *SQLiteDriver) SelectForUpdateSkipLocked() string

type SQLiteStorage 

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

SQLiteStorage implements the Storage interface using SQLite.

func NewSQLiteStorage 

func NewSQLiteStorage(dbPath string) (*SQLiteStorage, error)

NewSQLiteStorage creates a new SQLite storage.

func (*SQLiteStorage) AcquireLease 

func (s *SQLiteStorage) AcquireLease(ctx context.Context, id, owner string, ttl time.Duration) (Fence, bool, error)

AcquireLease grants a fresh epoch to a runnable unowned instance or an instance with an expired lease.

func (*SQLiteStorage) AppendSignal 

func (s *SQLiteStorage) AppendSignal(ctx context.Context, instanceID, channel string, payload []byte) (int64, error)

AppendSignal buffers a signal only while the instance is running.

func (*SQLiteStorage) ArrivedUndelivered 

func (s *SQLiteStorage) ArrivedUndelivered(ctx context.Context, id string) (*DeliveryBatch, error)

func (*SQLiteStorage) BeginTransaction 

func (s *SQLiteStorage) BeginTransaction(ctx context.Context) (context.Context, error)

BeginTransaction starts a new transaction.

func (*SQLiteStorage) CancelActivityAtYield 

func (s *SQLiteStorage) CancelActivityAtYield(ctx context.Context, f *Fence, id, key, cmd string, data []byte) (*HistoryEvent, error)

CancelActivityAtYield records cancellation and cancels the associated task atomically.

func (*SQLiteStorage) CancelChild 

func (s *SQLiteStorage) CancelChild(ctx context.Context, f *Fence, c *ChildCancel) (*HistoryEvent, error)

CancelChild resolves a handle and requests cancellation atomically.

func (*SQLiteStorage) ClaimActivityTask 

func (s *SQLiteStorage) ClaimActivityTask(ctx context.Context, id, key, owner string, ttl time.Duration, queues []string) (*ActivityTask, error)

ClaimActivityTask delegates to the durable task store.

func (*SQLiteStorage) ClaimActivityTasks 

func (s *SQLiteStorage) ClaimActivityTasks(ctx context.Context, owner string, ttl time.Duration, limit int, filter TaskFilter) ([]*ActivityTask, error)

ClaimActivityTasks delegates to the durable task store.

func (*SQLiteStorage) CleanupExpiredSystemLocks 

func (s *SQLiteStorage) CleanupExpiredSystemLocks(ctx context.Context) error

CleanupExpiredSystemLocks removes expired system locks.

func (*SQLiteStorage) Close 

func (s *SQLiteStorage) Close() error

Close closes the database connection.

func (*SQLiteStorage) CloseInstance 

func (s *SQLiteStorage) CloseInstance(ctx context.Context, f Fence, status WorkflowStatus, output []byte, event *HistoryEvent) error

CloseInstance commits a terminal status, result, close event, and timer cleanup together.

func (*SQLiteStorage) CommitObservation 

func (s *SQLiteStorage) CommitObservation(ctx context.Context, f *Fence, c *ObservationCommit) (*ObservationResult, error)

func (*SQLiteStorage) CommitTransaction 

func (s *SQLiteStorage) CommitTransaction(ctx context.Context) error

CommitTransaction commits the current transaction.

func (*SQLiteStorage) CompleteActivityTask 

func (s *SQLiteStorage) CompleteActivityTask(ctx context.Context, claim TaskClaim, event HistoryEventType, data []byte) error

CompleteActivityTask delegates to the durable task store.

func (*SQLiteStorage) Conn 

func (s *SQLiteStorage) Conn(ctx context.Context) Executor

Conn returns the database executor for the current context. If a transaction is active, returns the transaction; otherwise, returns the database. This allows storage operations to share an explicitly opened transaction.

Example: 

conn := store.Conn(ctx)
_, err := conn.ExecContext(ctx, "INSERT INTO orders (id) VALUES (?)", orderID)

func (*SQLiteStorage) CreateInstance 

func (s *SQLiteStorage) CreateInstance(ctx context.Context, instance *WorkflowInstance) error

CreateInstance creates a new workflow instance.

func (*SQLiteStorage) DB 

func (s *SQLiteStorage) DB() *sql.DB

DB returns the underlying database connection.

func (*SQLiteStorage) DeliveredSignals 

func (s *SQLiteStorage) DeliveredSignals(ctx context.Context, id string) ([]SignalPosition, error)

func (*SQLiteStorage) FindActivityTimeouts 

func (s *SQLiteStorage) FindActivityTimeouts(ctx context.Context, limit int) ([]*ActivityTimeout, error)

FindActivityTimeouts finds eligible timeout candidates.

func (*SQLiteStorage) FindDispatchable 

func (s *SQLiteStorage) FindDispatchable(ctx context.Context, limit int, workflowNames ...string) ([]string, error)

FindDispatchable lists runnable unowned instances and instances with expired leases.

func (*SQLiteStorage) FindExpiredTimers 

func (s *SQLiteStorage) FindExpiredTimers(ctx context.Context, limit int) ([]*TimerSubscription, error)

FindExpiredTimers finds expired timers.

func (*SQLiteStorage) FireTimer 

func (s *SQLiteStorage) FireTimer(ctx context.Context, timer *TimerSubscription) error

FireTimer atomically records a timer result and wake, or discards a terminal instance's timer.

func (*SQLiteStorage) GetHistoryCount 

func (s *SQLiteStorage) GetHistoryCount(ctx context.Context, instanceID string) (int64, error)

GetHistoryCount returns the total number of history events for an instance.

func (*SQLiteStorage) GetHistoryEvent 

func (s *SQLiteStorage) GetHistoryEvent(ctx context.Context, instanceID, key string) (*HistoryEvent, error)

GetHistoryEvent returns the stored row, or nil when absent.

func (*SQLiteStorage) GetHistoryPaginated 

func (s *SQLiteStorage) GetHistoryPaginated(ctx context.Context, instanceID string, afterID int64, limit int) ([]*HistoryEvent, bool, error)

GetHistoryPaginated retrieves history events with pagination. Returns events with id > afterID, up to limit events. Returns (events, hasMore, error).

func (*SQLiteStorage) GetInstance 

func (s *SQLiteStorage) GetInstance(ctx context.Context, instanceID string) (*WorkflowInstance, error)

GetInstance retrieves a workflow instance by ID.

func (*SQLiteStorage) InTransaction 

func (s *SQLiteStorage) InTransaction(ctx context.Context) bool

InTransaction returns whether a transaction is in progress.

func (*SQLiteStorage) Initialize 

func (s *SQLiteStorage) Initialize(ctx context.Context) error

Initialize is deprecated. Use dbmate for migrations: dbmate -d internal/migrations/sql/sqlite up

func (*SQLiteStorage) IsCancelRequested 

func (s *SQLiteStorage) IsCancelRequested(ctx context.Context, id string) (bool, error)

IsCancelRequested reads the durable cancellation request.

func (*SQLiteStorage) ListActivityTasks 

func (s *SQLiteStorage) ListActivityTasks(ctx context.Context, id string) ([]*ActivityTask, error)

ListActivityTasks returns tasks in reservation order.

func (*SQLiteStorage) ListInstances 

func (s *SQLiteStorage) ListInstances(ctx context.Context, opts ListInstancesOptions) (*PaginationResult, error)

ListInstances lists workflow instances with cursor-based pagination and filters.

func (*SQLiteStorage) MarkRunnable 

func (s *SQLiteStorage) MarkRunnable(ctx context.Context, id string) error

MarkRunnable marks a running instance for dispatch without changing ownership.

func (*SQLiteStorage) NextActivityDue 

func (s *SQLiteStorage) NextActivityDue(ctx context.Context, f TaskFilter) (time.Duration, bool, error)

NextActivityDue returns the delay until the next eligible queued task.

func (*SQLiteStorage) NextSignal 

func (s *SQLiteStorage) NextSignal(ctx context.Context, instanceID, channel string) (*Signal, error)

NextSignal returns the oldest unconsumed signal for this instance and channel.

func (*SQLiteStorage) NotifyRunnable 

func (s *SQLiteStorage) NotifyRunnable(ctx context.Context)

NotifyRunnable wakes Apps on this database in the current process.

func (*SQLiteStorage) ReadSnapshot 

func (s *SQLiteStorage) ReadSnapshot(ctx context.Context, id string) (*InstanceSnapshot, error)

ReadSnapshot uses a deferred transaction, bypassing the pool's immediate write transactions. In-memory SQLite databases (":memory:" with shared cache) cannot use WAL. Their snapshots block writers until the read transaction ends.

func (*SQLiteStorage) RecordHistory 

func (s *SQLiteStorage) RecordHistory(ctx context.Context, f *Fence, ev *HistoryEvent) (*HistoryEvent, error)

RecordHistory returns the first stored command and value at the key.

func (*SQLiteStorage) RegisterPostCommitCallback 

func (s *SQLiteStorage) RegisterPostCommitCallback(ctx context.Context, cb func() error) error

RegisterPostCommitCallback registers a callback to be executed after a successful commit. Returns an error if not currently in a transaction.

func (*SQLiteStorage) ReleaseActivityTask 

func (s *SQLiteStorage) ReleaseActivityTask(ctx context.Context, claim TaskClaim) error

ReleaseActivityTask delegates to the durable task store.

func (*SQLiteStorage) ReleaseLease 

func (s *SQLiteStorage) ReleaseLease(ctx context.Context, f Fence, runnable bool) error

ReleaseLease clears a matching lease and preserves pending input; runnable also requests another dispatch.

func (*SQLiteStorage) ReleaseSystemLock 

func (s *SQLiteStorage) ReleaseSystemLock(ctx context.Context, lockName, workerID string) error

ReleaseSystemLock releases a system lock.

func (*SQLiteStorage) RemoveTimerSubscription 

func (s *SQLiteStorage) RemoveTimerSubscription(ctx context.Context, instanceID, timerID string) error

RemoveTimerSubscription removes a timer subscription.

func (*SQLiteStorage) RenewActivityTask 

func (s *SQLiteStorage) RenewActivityTask(ctx context.Context, claim TaskClaim, ttl time.Duration, extend bool) error

RenewActivityTask delegates to the durable task store.

func (*SQLiteStorage) RenewLease 

func (s *SQLiteStorage) RenewLease(ctx context.Context, f Fence, ttl time.Duration) (bool, error)

RenewLease extends an unexpired matching lease using the store clock.

func (*SQLiteStorage) RequestCancel 

func (s *SQLiteStorage) RequestCancel(ctx context.Context, id, reason string) (WorkflowStatus, error)

RequestCancel requests cancellation without acquiring an execution lease.

func (*SQLiteStorage) ReserveActivity 

func (s *SQLiteStorage) ReserveActivity(ctx context.Context, f *Fence, seq int64, key, cmd string, task *ActivityTask) (*HistoryEvent, error)

ReserveActivity delegates to the durable task store.

func (*SQLiteStorage) ReserveHistory 

func (s *SQLiteStorage) ReserveHistory(ctx context.Context, f *Fence, instanceID, key, cmd string, seq int64) (*HistoryEvent, error)

ReserveHistory reserves a command key with a NULL value.

func (*SQLiteStorage) ReserveTimer 

func (s *SQLiteStorage) ReserveTimer(ctx context.Context, f *Fence, seq int64, timer *TimerSubscription) (*HistoryEvent, error)

ReserveTimer reserves a sleep and arms its timer atomically under the execution fence.

func (*SQLiteStorage) RetryActivityTask 

func (s *SQLiteStorage) RetryActivityTask(ctx context.Context, c TaskClaim, data []byte, delay time.Duration) error

RetryActivityTask durably schedules another attempt.

func (*SQLiteStorage) RollbackTransaction 

func (s *SQLiteStorage) RollbackTransaction(ctx context.Context) error

RollbackTransaction rolls back the current transaction. Callbacks are not executed on rollback.

func (*SQLiteStorage) SetHistoryValueIfNull 

func (s *SQLiteStorage) SetHistoryValueIfNull(ctx context.Context, f *Fence, instanceID, key, cmd string, t HistoryEventType, data []byte) (*HistoryEvent, error)

SetHistoryValueIfNull records the first value for a matching command.

func (*SQLiteStorage) StartChild 

func (s *SQLiteStorage) StartChild(ctx context.Context, f *Fence, seq int64, key, cmd string, child *WorkflowInstance) (*HistoryEvent, error)

StartChild reserves the parent command and creates its child in one fenced transaction.

func (*SQLiteStorage) TimeoutActivityTask 

func (s *SQLiteStorage) TimeoutActivityTask(ctx context.Context, c *ActivityTimeout, data []byte) error

TimeoutActivityTask atomically records a final timeout.

func (*SQLiteStorage) TryAcquireSystemLock 

func (s *SQLiteStorage) TryAcquireSystemLock(ctx context.Context, lockName, workerID string, ttl time.Duration) (bool, error)

TryAcquireSystemLock attempts to acquire a system lock.

func (*SQLiteStorage) UpdateInstanceActivity 

func (s *SQLiteStorage) UpdateInstanceActivity(ctx context.Context, instanceID, activityID string) error

UpdateInstanceActivity updates the current activity ID.

type ScopeCancel 

type ScopeCancel struct{ Targets []ScopeTarget }

type ScopeTarget 

type ScopeTarget struct{ Key, Command, Kind string }

ScopeTarget identifies an already issued activity, timer, or child.

type Signal 

type Signal struct {
	ID                        int64
	InstanceID, Channel       string
	Payload                   []byte
	SentAt                    time.Time
	ConsumedKey               string
	DeliveredSeq, ConsumedSeq int64
}

Signal is one durable inbox entry. An empty ConsumedKey denotes a pending entry.

type SignalConsume 

type SignalConsume struct {
	ID                     int64
	Channel                string
	RequireDeliveredBefore int64
}

type SignalManager 

type SignalManager interface {
	AppendSignal(ctx context.Context, instanceID, channel string, payload []byte) (int64, error)
	// NextSignal returns nil, nil when this channel has no unconsumed signal.
	NextSignal(ctx context.Context, instanceID, channel string) (*Signal, error)
}

SignalManager persists and consumes the per-instance signal inbox.

type SignalPosition 

type SignalPosition struct {
	ID                        int64
	Channel                   string
	DeliveredSeq, ConsumedSeq int64
}

type SignalRef 

type SignalRef struct {
	ID      int64
	Channel string
}

type SleepReservation 

type SleepReservation struct {
	ExpiresAt time.Time `json:"expires_at"`
}

SleepReservation holds immutable command metadata; EventData holds the winning observation.

type SnapshotReader 

type SnapshotReader interface {
	ReadSnapshot(context.Context, string) (*InstanceSnapshot, error)
}

SnapshotReader reads replay inputs without acquiring the workflow lease.

type Storage 

type Storage interface {
	// Initialize sets up the storage (create tables, etc.)
	Initialize(ctx context.Context) error

	// NotifyRunnable wakes dispatchers after committed runnable mutations.
	// Backends without cross-worker notifications implement a no-op.
	NotifyRunnable(context.Context)

	// Close closes the storage connection
	Close() error

	// DB returns the underlying database connection.
	// This is primarily used for migrations.
	DB() *sql.DB

	// Transaction Management
	TransactionManager

	// Workflow Instance Operations
	InstanceManager

	// Lease operations
	LeaseManager

	// History Operations
	HistoryManager

	DeliveryManager

	ActivityTaskManager
	ChildManager
	SnapshotReader

	// Timer Subscription Operations
	TimerSubscriptionManager

	// Per-instance signal inbox
	SignalManager

	// System Lock Operations
	SystemLockManager
}

Storage defines the interface for workflow persistence. Implementations must be safe for concurrent use.

type SystemLock 

type SystemLock struct {
	LockName  string    `json:"lock_name"`
	LockedBy  string    `json:"locked_by"`
	LockedAt  time.Time `json:"locked_at"`
	ExpiresAt time.Time `json:"expires_at"`
}

SystemLock represents a system-level lock for background tasks.

type SystemLockManager 

type SystemLockManager interface {
	// TryAcquireSystemLock attempts to acquire a system lock.
	// Returns true if the lock was acquired.
	TryAcquireSystemLock(ctx context.Context, lockName, workerID string, ttl time.Duration) (bool, error)

	// ReleaseSystemLock releases a system lock.
	ReleaseSystemLock(ctx context.Context, lockName, workerID string) error

	// CleanupExpiredSystemLocks removes expired system locks.
	CleanupExpiredSystemLocks(ctx context.Context) error
}

SystemLockManager handles system-level locks for background tasks.

type TaskClaim 

type TaskClaim struct {
	TaskID                        int64
	InstanceID, ActivityID, Owner string
	Epoch                         int64
}

TaskClaim identifies one activity attempt independently of the workflow lease.

type TaskFilter 

type TaskFilter struct{ Names, Queues []string }

TaskFilter restricts claims to registered names and optionally served queues. Nil Queues serves all queues; an empty non-nil slice serves none.

type TimerRetire 

type TimerRetire struct{ Key string }

type TimerSubscription 

type TimerSubscription struct {
	ID         int64     `json:"id"`
	InstanceID string    `json:"instance_id"`
	TimerID    string    `json:"timer_id"`
	ExpiresAt  time.Time `json:"expires_at"`
	ActivityID string    `json:"activity_id"` // Activity ID for replay matching
	CreatedAt  time.Time `json:"created_at"`
}

TimerSubscription represents a workflow waiting for a timer.

type TimerSubscriptionManager 

type TimerSubscriptionManager interface {
	// ReserveTimer atomically reserves a sleep and arms its timer with the first expiry.
	ReserveTimer(context.Context, *Fence, int64, *TimerSubscription) (*HistoryEvent, error)
	// FireTimer atomically records a timer result, wakes its instance, and removes the subscription.
	FireTimer(context.Context, *TimerSubscription) error

	// FindExpiredTimers finds timers that have expired.
	// limit specifies the maximum number of timers to return (0 = default 100).
	FindExpiredTimers(ctx context.Context, limit int) ([]*TimerSubscription, error)

	// RemoveTimerSubscription removes a timer subscription.
	RemoveTimerSubscription(ctx context.Context, instanceID, timerID string) error
}

TimerSubscriptionManager handles timer subscriptions.

type TransactionManager 

type TransactionManager interface {
	// BeginTransaction starts a new transaction.
	// Returns a context with the transaction attached.
	BeginTransaction(ctx context.Context) (context.Context, error)

	// CommitTransaction commits the current transaction.
	CommitTransaction(ctx context.Context) error

	// RollbackTransaction rolls back the current transaction.
	RollbackTransaction(ctx context.Context) error

	// InTransaction returns true if there is an active transaction.
	InTransaction(ctx context.Context) bool

	// Conn returns the database executor for the current context.
	// If a transaction is active, returns the transaction; otherwise, returns the database.
	// This allows users to execute custom SQL queries within the same transaction.
	Conn(ctx context.Context) Executor

	// RegisterPostCommitCallback registers a callback to be executed after a successful commit.
	// This is used for operations that should only happen if the transaction commits,
	// such as notifying waiting subscribers after a message is published.
	// Returns an error if not currently in a transaction.
	RegisterPostCommitCallback(ctx context.Context, cb func() error) error
}

TransactionManager handles transaction operations.

type WorkflowInstance 

type WorkflowInstance struct {
	ParentInstanceID  string         `json:"parent_instance_id,omitempty"`
	ParentKey         string         `json:"parent_key,omitempty"`
	CancelReason      string         `json:"cancel_reason,omitempty"`
	InstanceID        string         `json:"instance_id"`
	WorkflowName      string         `json:"workflow_name"`
	Framework         string         `json:"framework"` // "go" or "python"
	Status            WorkflowStatus `json:"status"`
	InputData         []byte         `json:"input_data"`          // JSON-encoded input
	OutputData        []byte         `json:"output_data"`         // JSON-encoded output (if completed)
	CurrentActivityID string         `json:"current_activity_id"` // Last executed activity ID
	SourceHash        string         `json:"source_hash"`         // Source code hash
	OwnerService      string         `json:"owner_service"`       // Service that owns this workflow
	LeaseOwner        string         `json:"lease_owner"`
	LeaseEpoch        int64          `json:"lease_epoch"`
	LeaseExpiresMS    *int64         `json:"lease_expires_ms"`
	Runnable          bool           `json:"runnable"`

	StartedAt time.Time `json:"started_at"` // When the workflow was created
	UpdatedAt time.Time `json:"updated_at"`
}

WorkflowInstance represents a single execution of a workflow.

type WorkflowStatus 

type WorkflowStatus string

WorkflowStatus represents the current state of a workflow instance. 

const (
	StatusRunning   WorkflowStatus = "running"
	StatusCompleted WorkflowStatus = "completed"
	StatusFailed    WorkflowStatus = "failed"
	StatusCancelled WorkflowStatus = "cancelled"
)

Source Files

View all Source files

Directories

Path Synopsis
Package faultstore provides storage fault injection for engine tests.
 Package faultstore provides storage fault injection for engine tests.

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.