drivertest

package
v0.0.8 Latest Latest
Warning

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

 Go to latest Published: Jul 31, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package drivertest provides an in-memory driver.Store used by the queue and event runtimes' unit tests. Fake honors the real contract semantics — durable budget resolution, lease-token fencing, reap-to-dead, idempotency dedupe and atomic publish fan-out — so runtime logic can be tested without a database.

It also implements the optional driver.Notifier, driver.LeaderElector and driver.ChangeNotifier capabilities so wake-driven fetch loops, leader-elected cron and change-hint streams can be exercised in memory. It deliberately does not implement driver.TxStore: the fake has no transactions.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Fake 

type Fake struct {
	// Clock is the time source; tests may replace it with a controllable clock
	// before use. It defaults to clock.SystemClock.
	Clock clock.Clock
	// contains filtered or unexported fields
}

Fake is a thread-safe, in-memory driver.Store for tests.

func NewFake 

func NewFake() *Fake

NewFake returns an empty Fake backed by the system clock.

func (*Fake) Ack 

func (f *Fake) Ack(_ context.Context, id, leaseToken uuid.UUID) error

Ack completes an active job, retaining it as succeeded.

func (*Fake) AckTaskResult 

func (f *Fake) AckTaskResult(_ context.Context, id, leaseToken uuid.UUID, result json.RawMessage) error

AckTaskResult completes an active task like Ack and persists its result atomically. Same lease-token fencing.

func (*Fake) AcquireLeadership 

func (f *Fake) AcquireLeadership(ctx context.Context, name string) (func(), bool, error)

AcquireLeadership takes the named in-memory lock if it is free.

func (*Fake) AcquireLeadershipLease added in v0.0.4 

func (f *Fake) AcquireLeadershipLease(_ context.Context, name string) (driver.LeadershipLease, bool, error)

AcquireLeadershipLease takes the named in-memory lock if it is free, returning a lease that InvalidateLeadership can later flip to lost — simulating a dropped session without an explicit Release, the way a real backend's connection can die out from under a held lock.

func (*Fake) AllDaily 

func (f *Fake) AllDaily(_ context.Context, source driver.Source) ([]driver.DailyCount, error)

AllDaily returns the daily throughput window summed across every kind.

func (*Fake) AppendHistory 

func (f *Fake) AppendHistory(_ context.Context, workflowID uuid.UUID, typ string, payload json.RawMessage) (int64, error)

AppendHistory appends one durable history record. For an OperationCompleted/OperationFailed payload carrying an execution_key already recorded, it is idempotent: it returns the existing record's seq instead of appending a duplicate — closing the crash window between AppendHistory and the operation job's Ack (processOperationJob).

func (*Fake) ApplyFailurePolicy 

func (f *Fake) ApplyFailurePolicy(_ context.Context) ([]driver.DAGFailure, error)

ApplyFailurePolicy applies each running workflow's OnFailure policy when a dead task triggers it. A dead task whose dependents all declare IgnoreDeadDeps does not trigger the policy (the ignoring branch keeps running); a dead task with no dependents always does.

func (*Fake) ArchiveJob 

func (f *Fake) ArchiveJob(_ context.Context, source driver.Source, id uuid.UUID) error

ArchiveJob force-fails a pending or scheduled job to dead.

func (*Fake) CancelDAG 

func (f *Fake) CancelDAG(_ context.Context, id uuid.UUID) error

CancelDAG cancels a non-terminal workflow without compensating. On a compensating workflow the not-yet-started compensations are cancelled, the origin is recorded, and CompleteDAGs lands the workflow on cancelled once the in-flight compensations settle; otherwise the workflow lands on cancelled immediately.

func (*Fake) CancelWorkflowExecution 

func (f *Fake) CancelWorkflowExecution(_ context.Context, id uuid.UUID) error

CancelWorkflowExecution settles a non-terminal workflow as cancelled.

func (*Fake) Changes added in v0.0.8 

func (f *Fake) Changes(ctx context.Context) (<-chan driver.Change, error)

Changes returns a stream of row-change hints; the channel is closed when ctx ends. Per the capability contract, the first delivery is a ChangeReset.

func (*Fake) Close 

func (f *Fake) Close(context.Context) error

Close is a no-op for the in-memory fake.

func (*Fake) CompensateDAG 

func (f *Fake) CompensateDAG(_ context.Context, id uuid.UUID) error

CompensateDAG manually triggers compensation on a running or suspended workflow, exactly like the cancel policy.

func (*Fake) CompleteDAGs 

func (f *Fake) CompleteDAGs(_ context.Context) (int64, error)

CompleteDAGs settles finished dags: a running workflow whose tasks are all terminal (succeeded or dead) lands succeeded when none died, or failed (recording the dead task keys) when a tolerated task died; a compensating workflow settles once its compensation tasks resolve — suspended on a dead compensation (unless the compensation was cancelled through CancelDAG), failed when the chain finished, or cancelled when it was triggered through CancelDAG.

func (*Fake) CompleteDueSleeps 

func (f *Fake) CompleteDueSleeps(_ context.Context) (int64, error)

CompleteDueSleeps succeeds every due $sleep timer of a running workflow; no handler is involved.

func (*Fake) CompleteWorkflow 

func (f *Fake) CompleteWorkflow(_ context.Context, id uuid.UUID, result json.RawMessage) error

CompleteWorkflow settles a workflow as succeeded, persisting the result.

func (*Fake) CreateDAG 

func (f *Fake) CreateDAG(_ context.Context, p driver.DAGParams) (bool, uuid.UUID, error)

CreateDAG atomically inserts the header, tasks and deps, deduplicating by (Name, IdempotencyKey) against live executions.

func (*Fake) DAGDeps added in v0.0.7 

func (f *Fake) DAGDeps(_ context.Context, id uuid.UUID) ([]driver.DAGDep, error)

DAGDeps returns the workflow's edges, compensation-chain links included, ordered by (task_key, depends_on_key). An unknown id yields no rows.

func (*Fake) DAGNameStateCounts added in v0.0.7 

func (f *Fake) DAGNameStateCounts(_ context.Context) (map[string]map[driver.DAGState]int64, error)

DAGNameStateCounts counts every retained workflow by (definition, state).

func (*Fake) DAGTaskCounts added in v0.0.7 

func (f *Fake) DAGTaskCounts(_ context.Context, ids []uuid.UUID) (map[uuid.UUID]map[driver.JobState]int64, error)

DAGTaskCounts breaks each requested workflow down by task state. Ids with no tasks are absent from the result.

func (*Fake) DAGTasks 

func (f *Fake) DAGTasks(_ context.Context, id uuid.UUID) ([]driver.Job, error)

DAGTasks returns the workflow's task jobs in creation order.

func (*Fake) Dead 

func (f *Fake) Dead(_ context.Context, id, leaseToken uuid.UUID, lastError string) error

Dead moves a failed active job to the dead-letter state and records the attempt.

func (*Fake) DeleteAll 

func (f *Fake) DeleteAll(_ context.Context, source driver.Source, kind string, state driver.JobState) (int64, error)

DeleteAll deletes every job of (source, kind) in the given state.

func (*Fake) DeleteJob 

func (f *Fake) DeleteJob(_ context.Context, source driver.Source, id uuid.UUID, state driver.JobState) error

DeleteJob deletes a job of the source in the given state.

func (*Fake) DeleteSubscriber added in v0.0.4 

func (f *Fake) DeleteSubscriber(_ context.Context, name, eventType string) (int64, error)

DeleteSubscriber removes the (name, eventType) registration, or every registration of name when eventType is empty.

func (*Fake) DeliverBufferedSignals added in v0.0.6 

func (f *Fake) DeliverBufferedSignals(_ context.Context) (int64, error)

DeliverBufferedSignals hands buffered inbox signals to tasks that became deliverable after the signal arrived: per task, the oldest unconsumed signal of its name wins. Set-based and idempotent, like the SQL statement.

func (*Fake) DequeueBatch 

func (f *Fake) DequeueBatch(_ context.Context, source driver.Source, p driver.DequeueParams) ([]driver.Job, error)

DequeueBatch leases up to Limit due pending jobs of (source, Kind).

func (*Fake) Enqueue 

func (f *Fake) Enqueue(_ context.Context, p driver.EnqueueParams) (bool, error)

Enqueue inserts one queue job, honoring both idempotency modes.

func (*Fake) ExtendLease 

func (f *Fake) ExtendLease(_ context.Context, id, leaseToken uuid.UUID, lease time.Duration) error

ExtendLease renews an active job's lease.

func (*Fake) FailWorkflow 

func (f *Fake) FailWorkflow(_ context.Context, id uuid.UUID, reason string) error

FailWorkflow settles a workflow as failed, recording the reason.

func (*Fake) FindDAGByKey added in v0.0.6 

func (f *Fake) FindDAGByKey(_ context.Context, name, idempotencyKey string) (uuid.UUID, error)

FindDAGByKey resolves the live workflow holding (name, idempotencyKey).

func (*Fake) GetDAG 

func (f *Fake) GetDAG(_ context.Context, id uuid.UUID) (*driver.DAGView, error)

GetDAG returns one workflow header by id.

func (*Fake) GetEvent 

func (f *Fake) GetEvent(_ context.Context, id uuid.UUID) (*driver.EventAdminRow, error)

GetEvent returns one ledger event by id, or a not-found error.

func (*Fake) GetJob 

func (f *Fake) GetJob(_ context.Context, source driver.Source, id uuid.UUID) (*driver.Job, error)

GetJob returns one job of the source by id, or a not-found error.

func (*Fake) GetWorkflowExecution 

func (f *Fake) GetWorkflowExecution(_ context.Context, id uuid.UUID) (driver.WorkflowExecutionView, error)

GetWorkflowExecution returns one execution header by id, or a not-found error when it does not exist.

func (*Fake) InvalidateLeadership added in v0.0.4 

func (f *Fake) InvalidateLeadership(name string)

InvalidateLeadership simulates the named leadership's backing session dying: the next Valid check on its lease reports false, without releasing the in-memory lock (mirroring a real advisory lock outliving the caller's awareness until it re-checks). A name with no active lease is a no-op.

func (*Fake) JobAttempts 

func (f *Fake) JobAttempts(_ context.Context, source driver.Source, id uuid.UUID) ([]driver.AttemptError, error)

JobAttempts returns a job's failure history, oldest first.

func (*Fake) KindDepths 

func (f *Fake) KindDepths(_ context.Context, source driver.Source) (map[string]driver.Depths, error)

KindDepths returns per-kind state counters of the source.

func (*Fake) ListDAGs 

func (f *Fake) ListDAGs(_ context.Context, filter driver.DAGFilter, offset, limit int) ([]driver.DAGView, int64, error)

ListDAGs lists dags matching the filter, newest first, paginated.

func (*Fake) ListEvents 

func (f *Fake) ListEvents(_ context.Context, filter driver.EventFilter, offset, limit int) ([]driver.EventAdminRow, int64, error)

ListEvents lists ledger events matching the filter, newest first, paginated.

func (*Fake) ListHistory 

func (f *Fake) ListHistory(_ context.Context, workflowID uuid.UUID) ([]driver.HistoryEvent, error)

ListHistory returns the workflow's history events in sequence order.

func (*Fake) ListJobs 

func (f *Fake) ListJobs(_ context.Context, source driver.Source, filter driver.JobFilter, offset, limit int) ([]driver.Job, int64, error)

ListJobs lists jobs of the source matching the filter, paginated.

func (*Fake) ListKinds 

func (f *Fake) ListKinds(_ context.Context, source driver.Source) ([]string, error)

ListKinds returns the distinct kinds of the source, sorted.

func (*Fake) ListStalledWorkflows added in v0.0.4 

func (f *Fake) ListStalledWorkflows(_ context.Context, olderThan time.Duration, limit int) ([]driver.StalledWorkflow, error)

ListStalledWorkflows returns up to limit running executions with no live (pending, scheduled, active or uncertain) source=workflow job, updated at least olderThan ago. The cutoff is inclusive (After, not !Before): on coarse clocks (Windows ticks at up to ~15.6ms) an execution created and checked within one tick has UpdatedAt == cutoff at olderThan 0, and a strict comparison would hide a genuinely stranded execution.

func (*Fake) ListSubscriberViews 

func (f *Fake) ListSubscriberViews(_ context.Context, eventType string) ([]driver.SubscriberView, error)

ListSubscriberViews returns subscriber registrations, ordered by type then name.

func (*Fake) MarkUncertain 

func (f *Fake) MarkUncertain(_ context.Context, operationJobID, leaseToken uuid.UUID, reason string) error

MarkUncertain moves an active Operation to StateUncertain and suspends the parent execution.

func (*Fake) NukeAll 

func (f *Fake) NukeAll(_ context.Context, source driver.Source) (driver.NukeReport, error)

NukeAll deletes all jobs, stats and idempotency keys of the source.

func (*Fake) OpsStats 

func (f *Fake) OpsStats(_ context.Context) (driver.OpsStats, error)

OpsStats returns the ledger admin summary.

func (*Fake) PauseDAG added in v0.0.6 

func (f *Fake) PauseDAG(_ context.Context, id uuid.UUID, reason string) error

PauseDAG freezes a running workflow: header to paused with the reason recorded, pending/scheduled tasks to paused. Blocked and waiting tasks keep their states; nothing promotes while paused.

func (*Fake) PauseJob 

func (f *Fake) PauseJob(_ context.Context, source driver.Source, id uuid.UUID) error

PauseJob holds a pending or scheduled job out of the ready set.

func (*Fake) PromoteDue 

func (f *Fake) PromoteDue(_ context.Context, source driver.Source, kinds []string) (int64, error)

PromoteDue moves due scheduled jobs of the kinds to pending.

func (*Fake) PromoteUnblocked 

func (f *Fake) PromoteUnblocked(_ context.Context) (int64, error)

PromoteUnblocked releases blocked tasks whose dependencies are all satisfied, into the runnable state their kind dictates. Only running and compensating dags promote: a suspended workflow starts nothing new until an operator verb resumes it.

func (*Fake) Publish 

func (f *Fake) Publish(_ context.Context, p driver.PublishParams) (int, error)

Publish appends the event and fans out one pending delivery per subscriber.

func (*Fake) ReapExpired 

func (f *Fake) ReapExpired(_ context.Context, source driver.Source, kinds []string, maxReaps int) (int64, int64, error)

ReapExpired reclaims active jobs whose lease expired, killing poison jobs.

func (*Fake) RegisterSubscriber 

func (f *Fake) RegisterSubscriber(_ context.Context, sub driver.Subscriber) error

RegisterSubscriber upserts a subscriber keyed by (name, event type).

func (*Fake) Release 

func (f *Fake) Release(_ context.Context, id, leaseToken uuid.UUID) error

Release returns a leased job to pending without recording an attempt, decrementing the attempt it did not really spend.

func (*Fake) Replay 

func (f *Fake) Replay(_ context.Context, filter driver.ReplayFilter) (int64, error)

Replay re-fans-out ledger events matching the filter into fresh deliveries.

func (*Fake) Reschedule 

func (f *Fake) Reschedule(_ context.Context, id, leaseToken uuid.UUID, delay time.Duration, lastError string) error

Reschedule parks a failed active job for a later retry and records the attempt.

func (*Fake) ResolveUncertain 

func (f *Fake) ResolveUncertain(_ context.Context, operationJobID uuid.UUID, decision string, result json.RawMessage) (uuid.UUID, string, error)

ResolveUncertain applies complete/fail/retry to an uncertain Operation. History append is the caller's responsibility.

func (*Fake) ResumeJob 

func (f *Fake) ResumeJob(_ context.Context, source driver.Source, id uuid.UUID) error

ResumeJob returns a paused job to pending or scheduled per its run_at.

func (*Fake) ResumeWorkflow 

func (f *Fake) ResumeWorkflow(_ context.Context, id uuid.UUID) error

ResumeWorkflow moves a suspended execution back to running.

func (*Fake) Retain 

func (f *Fake) Retain(_ context.Context, before time.Time, limit int) (int64, error)

Retain deletes old events whose deliveries have all reached a terminal state, cascading to those terminal deliveries.

func (*Fake) RetryAllDead 

func (f *Fake) RetryAllDead(_ context.Context, source driver.Source, kind string) (int64, error)

RetryAllDead resets every dead job of (source, kind) to pending.

func (*Fake) RetryDAG 

func (f *Fake) RetryDAG(_ context.Context, id uuid.UUID) error

RetryDAG resumes a non-terminal workflow after failures. On a workflow with a compensation chain only the dead compensation tasks are reset and the workflow resumes compensating (original tasks never rerun once compensation started); otherwise every dead task is reset with a fresh budget and a suspended workflow resumes running.

func (*Fake) RetryJob 

func (f *Fake) RetryJob(_ context.Context, source driver.Source, id uuid.UUID) error

RetryJob resets a dead job of the source to pending.

func (*Fake) RunNow added in v0.0.6 

func (f *Fake) RunNow(_ context.Context, source driver.Source, id uuid.UUID) error

RunNow expedites a scheduled job to pending with run_at = now.

func (*Fake) ScheduleOperation 

func (f *Fake) ScheduleOperation(_ context.Context, p driver.ScheduleOperationParams) (uuid.UUID, error)

ScheduleOperation inserts one Operation task job, deduping by ExecutionKey among non-terminal workflow jobs.

func (*Fake) ScheduleTask 

func (f *Fake) ScheduleTask(_ context.Context, workflowID uuid.UUID, kind string, runAt time.Time) error

ScheduleTask durably inserts one workflow-task job (Source SourceWorkflow, RunID = workflowID), born pending when runAt is due, scheduled otherwise — the same split Enqueue applies to queue jobs.

func (*Fake) SeedOrphanDelivery 

func (f *Fake) SeedOrphanDelivery(deliveryID uuid.UUID, subscriber string)

SeedOrphanDelivery inserts a pending event delivery job whose ledger event is absent, so a dequeue rehydrates it with a nil Event. It exists purely to let the event runtime's tests exercise the "delivery without a ledger record" dead-letter path, which Publish (always atomic with the ledger) cannot produce. deliveryID is the job's id; the missing event id is generated.

func (*Fake) Signal 

func (f *Fake) Signal(_ context.Context, p driver.DAGSignalParams) (int64, bool, error)

Signal delivers (or buffers) one named signal on a live workflow: dedupe by MessageID, immediate delivery to a waiting $signal / scheduled $sleep, and durable buffering otherwise — mirroring azyncpgx's single transaction.

func (*Fake) SignalWorkflow 

func (f *Fake) SignalWorkflow(_ context.Context, p driver.SignalParams) (bool, error)

SignalWorkflow appends an early signal to the inbox, deduplicating by (WorkflowID, Name, MessageID) when MessageID is set; an empty MessageID disables dedupe, so every such Signal is a distinct message. SignalWorkflow atomically appends the delivery to the inbox (deduped by MessageID), the SignalReceived history record, and — when the execution is not already terminal — a wake workflow-task job, in the same critical section. Matches azyncpgx's single-transaction contract: a newly delivered signal is never observed recorded without also being able to wake a parked replay (the pre-fix race this closes: a crash between the inbox insert and the wake-task schedule left a signal "delivered" with nothing to ever act on it).

func (*Fake) Skip added in v0.0.6 

func (f *Fake) Skip(_ context.Context, id, leaseToken uuid.UUID, reason string) error

Skip settles an active job as skipped: terminal deliberate no-op, the reason retained, counted as processed, no attempts row.

func (*Fake) Snooze 

func (f *Fake) Snooze(_ context.Context, id, leaseToken uuid.UUID, delay time.Duration, deadlineError string) (bool, error)

Snooze parks an active job as scheduled after delay without consuming the retry budget: the attempt of the lease being handed back is decremented and no attempt history is recorded (polling-wait semantics). A job with a snooze budget stamps its deadline on the first snooze; a snooze settled past a stamped deadline dead-letters the job instead (deadlined=true), recording the final attempt — mirroring azyncpgx's atomic escalation.

func (*Fake) StartWorkflow 

func (f *Fake) StartWorkflow(_ context.Context, p driver.WorkflowStartParams) (bool, uuid.UUID, error)

StartWorkflow atomically inserts one workflow-as-code execution header, deduplicating by (Name, BusinessIdempotencyKey) against live (running or suspended) executions, and — for a newly inserted execution — records the WorkflowStarted history event and schedules the first workflow-task job in the same critical section, matching azyncpgx's single-transaction contract: a caller never observes an execution with no history or task.

func (*Fake) Stats 

func (f *Fake) Stats(_ context.Context, source driver.Source, kind string) (driver.Depths, []driver.DailyCount, error)

Stats returns one kind's depths and daily throughput window.

func (*Fake) Subscribers 

func (f *Fake) Subscribers(_ context.Context, eventType string) ([]driver.Subscriber, error)

Subscribers returns the registrations for an event type, ordered by name.

func (*Fake) SuspendWorkflow 

func (f *Fake) SuspendWorkflow(_ context.Context, id uuid.UUID, reason string) error

SuspendWorkflow parks a running workflow for a manual decision.

func (*Fake) TaskResults 

func (f *Fake) TaskResults(_ context.Context, dagID uuid.UUID, keys []string) (map[string]driver.TaskResult, error)

TaskResults returns the persisted results of the workflow's succeeded tasks, keyed by task key; an empty keys slice selects every succeeded task.

func (*Fake) VacuumCompleted 

func (f *Fake) VacuumCompleted(_ context.Context, source driver.Source, retention time.Duration) (int64, error)

VacuumCompleted trims succeeded jobs of the source completed before retention ago. DAG-owned (DAGID != nil) and workflow-as-code (RunID != nil) jobs are exempt: cleaned up only by VacuumDAGs / VacuumWorkflows.

func (*Fake) VacuumDAGs 

func (f *Fake) VacuumDAGs(_ context.Context, retention time.Duration) (int64, error)

VacuumDAGs deletes terminal dags completed before retention ago, cascading to their task jobs, attempt history and dependency edges.

func (*Fake) VacuumDead 

func (f *Fake) VacuumDead(_ context.Context, source driver.Source, kind string, olderThan time.Duration) (int64, error)

VacuumDead deletes dead jobs of (source, kind) enqueued before olderThan ago.

func (*Fake) VacuumIdempotency 

func (f *Fake) VacuumIdempotency(_ context.Context, source driver.Source) (int64, error)

VacuumIdempotency trims expired time-window dedupe keys of the source.

func (*Fake) VacuumStats 

func (f *Fake) VacuumStats(_ context.Context, source driver.Source, retention time.Duration) (int64, error)

VacuumStats trims stat counters of the source older than retention.

func (*Fake) VacuumWorkflows 

func (f *Fake) VacuumWorkflows(_ context.Context, retention time.Duration) (int64, error)

VacuumWorkflows deletes terminal workflow-as-code executions completed before retention ago, cascading to their jobs and history.

func (*Fake) Wake 

func (f *Fake) Wake(ctx context.Context) (<-chan driver.Wake, error)

Wake returns a channel of wakeups; it is closed when ctx ends.

func (*Fake) WorkflowExecutionCount 

func (f *Fake) WorkflowExecutionCount() int

WorkflowExecutionCount returns the number of workflow-as-code executions currently held, for tests that assert on vacuum-style cleanup.

type ManualClock 

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

ManualClock is a thread-safe, manually advanced clock.Clock for tests that need to control the Fake's view of time (lease expiry, run_at promotion, retention cutoffs) without sleeping.

func NewManualClock 

func NewManualClock(start time.Time) *ManualClock

NewManualClock returns a ManualClock frozen at start.

func (*ManualClock) Advance 

func (c *ManualClock) Advance(d time.Duration)

Advance moves the clock forward by d.

func (*ManualClock) Now 

func (c *ManualClock) Now() time.Time

Now returns the clock's current frozen time.

func (*ManualClock) Set 

func (c *ManualClock) Set(t time.Time)

Set jumps the clock to t (backwards jumps are allowed, e.g. to simulate clock skew between instances).

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.