app

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Oct 5, 2026 License: Apache-2.0 Imports: 15 Imported by: 0

Documentation

Overview

Package app implements the bounded terminal-work processor and provider router (requirements 8.4–8.8, design D9).

Index

Constants

View Source 
const (
	MetricsPageSize       = 256
	MaxMetricsScanRecords = 100_000
)

Metrics page/scan bounds (requirement 8.9, 12.5–12.8).

View Source 
const (
	TransitionClaimFailed      = "claim_failed"
	TransitionValidationFailed = "validation_failed"
	TransitionAcquireFailed    = "acquire_failed"
)

Variables

View Source 
var (
	ErrInvalidPayload    = errors.New("terminalwork: invalid payload")
	ErrProviderTimeout   = errors.New("terminalwork: provider timeout")
	ErrProviderOutage    = errors.New("terminalwork: provider outage")
	ErrAmbiguousCommit   = errors.New("terminalwork: ambiguous commit")
	ErrMissingProvider   = errors.New("terminalwork: missing provider")
	ErrDuplicateProvider = errors.New("terminalwork: duplicate provider")
	ErrNilProvider       = errors.New("terminalwork: nil provider")
	ErrUnsupportedKind   = errors.New("terminalwork: unsupported work kind")
	ErrMalformedProvider = errors.New("terminalwork: malformed provider")
	ErrClaimRenewFailed  = errors.New("terminalwork: claim renew failed")
	ErrNotRunning        = errors.New("terminalwork: processor not running")
	ErrAlreadyStarted    = errors.New("terminalwork: processor already started")
	// ErrDurablePending signals that required terminal effects were not completed
	// live but durable intent was accepted (requirements 7.7, 8.3; design D9).
	ErrDurablePending = errors.New("terminalwork: durable pending")
	// ErrDurableIntentRejected signals settle/release failed and durable intent
	// could not be accepted — live state must stay incomplete (design D9).
	ErrDurableIntentRejected = errors.New("terminalwork: durable intent rejected")
	// ErrQueryTooBroad / ErrQueryUnsupported are operator query rejections (8.9, 12.6).
	ErrQueryTooBroad    = errors.New("terminalwork: query too broad")
	ErrQueryUnsupported = errors.New("terminalwork: query unsupported")
	ErrNilIntentStore   = errors.New("terminalwork: nil intent store")
	// ErrMetricsCursorFault / ErrMetricsBoundExceeded protect Snapshot pagination.
	ErrMetricsCursorFault   = errors.New("terminalwork: metrics cursor fault")
	ErrMetricsBoundExceeded = errors.New("terminalwork: metrics scan bound exceeded")
)

Classified invoke outcomes used by the processor (requirements 8.5, 8.7).

View Source 
var ErrAmbiguousAppendReconcilerNotConfigured = errors.New("terminalwork: ambiguous append reconciler not configured")

ErrAmbiguousAppendReconcilerNotConfigured means an ambiguous append occurred and no AmbiguousAppendHandoff was wired. The candidate pin is released; this does not claim durable no-drop (subtask B wires the reconciler).

View Source 
var ErrAppendReconcileAmbiguous = errors.New("terminalwork: append reconcile ambiguous")

ErrAppendReconcileAmbiguous means AppendIntent failed and lookup could not confirm absence or classify durable state for safe ownership.

View Source 
var ErrIntentReplayConflict = errors.New("terminalwork: intent replay conflict")

ErrIntentReplayConflict means an existing WorkID row conflicts with this accept.

Functions

func DecodeHandlesPayload 

func DecodeHandlesPayload(payload []byte) ([]string, error)

DecodeHandlesPayload extracts sorted-or-stored handles from durable intent JSON.

func DecodeLeaseSetPayload 

func DecodeLeaseSetPayload(rec terminalwork.WorkRecord) (setID, reason string, err error)

DecodeLeaseSetPayload extracts set identity from durable work.

func IsPermanent 

func IsPermanent(err error) bool

IsPermanent reports whether err must quarantine rather than retry. Missing/temporarily unresolvable providers are never permanent (design D9).

Types

type AdoptionToken 

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

AdoptionToken is an in-flight WorkID adoption reservation. BeginAdoption before append/replay lookup; End on every return path. Executable Bind must run only via PublishBound so winner selection and pending hold are atomic with MarkTerminal.

func (*AdoptionToken) End 

func (tok *AdoptionToken) End()

End releases the adoption reservation. When no reservations and no ownership remain, the WorkID slot (including a terminal tombstone) is removed so the tracker stays bounded by in-flight admission/ownership.

func (*AdoptionToken) Publish 

func (tok *AdoptionToken) Publish(pin PinRelease, clearExec func()) bool

Publish installs pre-built combined ownership under the reservation. Prefer PublishBound for production paths so executable Bind cannot race MarkTerminal. Rejects (and unwinds the candidate) when terminal was marked, a winner already exists, or the token is invalid/ended. Never clears a winner's resources.

func (*AdoptionToken) PublishBound 

func (tok *AdoptionToken) PublishBound(pin PinRelease, bind OwnershipBinder) bool

PublishBound atomically selects the tracker winner and invokes bind for that eligible candidate only. MarkTerminal cannot interleave between Bind and ownership publication. Losers release pin outside the lock and never call bind. Runtime-pin-only (bind creates nothing) and executable-only (pin=nil) ownership remain valid; reject only when both resulting resources are nil.

type AmbiguousAppend 

type AmbiguousAppend struct {
	WorkID string
	Record terminalwork.WorkRecord
	Pin    genpin.Pin
	Cause  error
}

AmbiguousAppend describes an append whose durable outcome is not known and must not be parked in GenerationPinTracker without a reconciliation owner.

type AmbiguousAppendHandoff 

type AmbiguousAppendHandoff interface {
	Take(ctx context.Context, amb AmbiguousAppend) error
}

AmbiguousAppendHandoff is the subtask-B seam that takes candidate ownership for asynchronous reconciliation. Tests may supply a fake that retains the pin.

type AmbiguousAppendReconciler 

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

AmbiguousAppendReconciler owns candidate runtime pins after an ambiguous append until durable state is definitive. Exactly one tracked worker processes a WorkID-keyed queue; no per-item goroutines. Memory stays bounded by Capacity; capacity waiters block in Take without an overflow queue or extra goroutines.

Ownership contract for Take:

  • Once Take receives an already-ambiguous append, queue-capacity backpressure is independent of request cancellation. A canceled or deadline-exceeded caller context never releases the candidate pin, never removes the WorkID, and never returns a cancellation error from the capacity wait. The blocked Take call remains the owner until capacity opens (ownership transfers to the process worker) or the same WorkID is already owned by the reconciler (safe duplicate/conflict resolution).
  • Before a successful Take return, the caller owns the candidate pin. Validation errors and ErrNotRunning release that pin exactly once. Production starts the reconciler before HTTP admission and shuts it down only after handlers/generations drain, so Take must not observe not-running while capacity-blocked; that release path is only for the impossible misconfiguration of calling Take when the reconciler is not accepting.
  • After a successful Take return, the reconciler owns the candidate pin until adoption, conflict, terminal classification, or definitive insert. Worker store operations use a private process context, not the request.

func NewAmbiguousAppendReconciler 

func NewAmbiguousAppendReconciler(store AmbiguousAppendStore, cfg AmbiguousAppendReconcilerConfig) (*AmbiguousAppendReconciler, error)

NewAmbiguousAppendReconciler constructs a reconciler. Start must be called before Take. Store and Pins are required.

func (*AmbiguousAppendReconciler) Pending 

func (r *AmbiguousAppendReconciler) Pending() int

Pending reports unique WorkIDs awaiting reconciliation (tests/readiness).

func (*AmbiguousAppendReconciler) SetExecutablePending 

func (r *AmbiguousAppendReconciler) SetExecutablePending(b ExecutablePendingBinder)

SetExecutablePending updates the optional executable pending binder.

func (*AmbiguousAppendReconciler) Shutdown 

func (r *AmbiguousAppendReconciler) Shutdown(ctx context.Context) error

Shutdown stops new handoffs and waits for the queue to drain. When the queue is empty, the worker stops and Shutdown succeeds. When items remain past ctx, returns ctx error without releasing pins or discarding items; the worker stays live and handoffs resume so a later recovery drain can complete.

func (*AmbiguousAppendReconciler) Start 

func (r *AmbiguousAppendReconciler) Start() error

Start owns exactly one worker goroutine until a successful empty Shutdown.

func (*AmbiguousAppendReconciler) Take 

func (r *AmbiguousAppendReconciler) Take(ctx context.Context, amb AmbiguousAppend) error

Take transfers candidate ownership into the reconciler queue.

Same-WorkID duplicate detection runs before capacity wait and releases the loser only when an existing reconciler item already owns that WorkID. Conflict likewise releases the loser. Queue-full waits ignore request cancellation: ctx is retained for AmbiguousAppendHandoff compatibility and is not consulted during capacity backpressure.

On ErrNotRunning / validation errors the still-caller-owned pin is released exactly once. After success the reconciler owns the pin until a definitive outcome.

type AmbiguousAppendReconcilerConfig 

type AmbiguousAppendReconcilerConfig struct {
	Capacity       int
	OperationLimit time.Duration
	RetryMin       time.Duration
	RetryMax       time.Duration
	Clock          func() time.Time
	// After waits for d unless ctx is canceled. Tests inject a barrier seam.
	After             func(ctx context.Context, d time.Duration) error
	Pins              *GenerationPinTracker
	ExecutablePending ExecutablePendingBinder
}

AmbiguousAppendReconcilerConfig configures the process-owned ambiguous-append worker.

type AmbiguousAppendStore 

type AmbiguousAppendStore interface {
	AppendIntentOutcome(context.Context, terminalwork.WorkRecord) (terminalwork.AppendIntentOutcome, error)
	LookupIntent(context.Context, string) (terminalwork.WorkRecord, bool, error)
	PromotePending(context.Context, terminalwork.PromotePendingCommand) error
}

AmbiguousAppendStore is the durable seam used by the process-owned reconciler.

type AppendIntentOutcomeStore 

type AppendIntentOutcomeStore interface {
	AppendIntentOutcome(ctx context.Context, rec terminalwork.WorkRecord) (terminalwork.AppendIntentOutcome, error)
}

AppendIntentOutcomeStore is the definitive production append seam. Inserted and Replay are mutually exclusive; ambiguous transport/store errors return neither flag with a non-nil error.

type AuthorityRequestEffectConfig 

type AuthorityRequestEffectConfig struct {
	ProviderID string
	Provider   authority.RequestProvider
	Version    string
}

AuthorityRequestEffectConfig wires a lipsdk RequestProvider as a terminal-work EffectProvider that decodes durable settle/release payloads.

type AuthorityRequestEffectProvider 

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

AuthorityRequestEffectProvider recovers settle/release via the live authority provider using handles persisted in the durable work payload.

func NewAuthorityRequestEffectProvider 

func NewAuthorityRequestEffectProvider(cfg AuthorityRequestEffectConfig) (*AuthorityRequestEffectProvider, error)

NewAuthorityRequestEffectProvider validates and returns an effect adapter.

func (*AuthorityRequestEffectProvider) Invoke 

func (p *AuthorityRequestEffectProvider) Invoke(ctx context.Context, rec terminalwork.WorkRecord, idempotencyKey string) error

Invoke decodes the durable handles payload and calls SettleRequest or ReleaseRequest.

func (*AuthorityRequestEffectProvider) ProviderID 

func (p *AuthorityRequestEffectProvider) ProviderID() string

func (*AuthorityRequestEffectProvider) SupportedKinds 

func (p *AuthorityRequestEffectProvider) SupportedKinds() []sdk.WorkKind

func (*AuthorityRequestEffectProvider) Version 

func (p *AuthorityRequestEffectProvider) Version() string

type Clock 

type Clock interface {
	Now() time.Time
}

Clock is an injectable time source for claim/retry decisions.

type Config 

type Config struct {
	OwnerID        string
	ClaimTTL       time.Duration
	ClaimLimit     int
	GlobalMax      int
	PerProviderMax int
	RetrySchedule  terminalwork.RetrySchedule
	Clock          Clock
	TickInterval   time.Duration
	RenewInterval  time.Duration
	TickC          <-chan struct{}
	RenewPulse     <-chan struct{}
	NewTicker      func(time.Duration) Ticker
	Metrics        ProcessMetrics
	// OnTerminalDone is invoked after successful Complete or Quarantine so
	// generation pending-provider refs can be cleared (requirement 9.8).
	// Prefer composing callbacks before Processor.Start; post-start registration
	// uses AddOnTerminalDone (synchronized, multi-subscriber).
	OnTerminalDone func(rec terminalwork.WorkRecord)
	// GenerationPins releases runtime-generation pins on terminal completion.
	GenerationPins *GenerationPinTracker
	// GenerationResolver resolves providers for rows with exact runtime
	// instance+generation identity. Nil falls back to process registry only for
	// legacy rows; exact/malformed rows fail closed when resolver is missing.
	GenerationResolver GenerationBoundResolver
}

type EffectProvider 

type EffectProvider interface {
	ProviderID() string
	SupportedKinds() []sdk.WorkKind
	Version() string
	Invoke(ctx context.Context, rec terminalwork.WorkRecord, idempotencyKey string) error
}

EffectProvider is a provider-neutral terminal-effect handler routed by stable ID.

type ExecutablePendingBinder 

type ExecutablePendingBinder interface {
	Bind(workID string, versions terminalwork.BoundVersions) (release func(), ok bool)
}

ExecutablePendingBinder registers WorkID-keyed executable-generation pending ownership without importing snapshotgen into terminalwork (task 3.6). Bind returns a release handle only when this call created the hold (ok=true). When AddPendingWork is idempotent no-op (ok=false), no release handle is returned so a loser candidate cannot release the winner's WorkID.

type FrozenTerminalProviders 

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

FrozenTerminalProviders is an immutable snapshot of terminal effect providers captured at generation compilation. Future candidates may differ without mutating prior generation views.

func SnapshotTerminalProviders 

func SnapshotTerminalProviders(reg *Registry) *FrozenTerminalProviders

SnapshotTerminalProviders copies providers from a Registry into an immutable view.

func (*FrozenTerminalProviders) Resolve 

func (v *FrozenTerminalProviders) Resolve(providerID string, kind sdk.WorkKind) (EffectProvider, error)

Resolve implements TerminalProviderView.

type GenerationBoundResolver 

type GenerationBoundResolver interface {
	Resolve(runtimeInstanceID, runtimeGenerationID, providerID string, kind sdk.WorkKind) (EffectProvider, error)
}

GenerationBoundResolver resolves effect providers from a retained runtime configuration generation's immutable request-plane/provider view. Explicit runtime-identity rows must fail closed when that exact instance+generation/provider cannot be resolved — never fall through to a newer process-global same-ID provider (task 3.6).

type GenerationPinTracker 

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

GenerationPinTracker holds process-owned runtime-generation pins keyed by durable terminal-work ID (req 10.3, 10.7). Each WorkID adopts at most one combined ownership entry containing an optional runtime pin and/or an optional executable-generation pending clear handle.

Adoption is ordered with terminal transitions via BeginAdoption reservations: MarkTerminal under the same mutex prevents late Publish after the sole terminal callback has run.

func NewGenerationPinTracker 

func NewGenerationPinTracker() *GenerationPinTracker

NewGenerationPinTracker returns an empty tracker bounded by pending work.

func (*GenerationPinTracker) AbortOwnership 

func (t *GenerationPinTracker) AbortOwnership(workID string)

AbortOwnership is the nonterminal abort seam: remove candidate ownership when durable append definitively did not happen. Alias of Release.

func (*GenerationPinTracker) AdoptHold 

func (t *GenerationPinTracker) AdoptHold(workID string, pin PinRelease, clearExec func()) bool

AdoptHold attempts Hold; on rejection the loser pin and clearExec (when provided by this candidate) are unwound. Never clears a winner hold.

func (*GenerationPinTracker) BeginAdoption 

func (t *GenerationPinTracker) BeginAdoption(workID string) *AdoptionToken

BeginAdoption reserves workID for adoption under the tracker mutex. The returned token must End on every path. Nil tracker / empty workID yield a no-op token.

func (*GenerationPinTracker) EntryCount 

func (t *GenerationPinTracker) EntryCount() int

EntryCount reports tracked WorkID slots including terminal tombstones and in-flight reservations (test/diagnostics for bounded cleanup).

func (*GenerationPinTracker) Hold 

func (t *GenerationPinTracker) Hold(workID string, pin PinRelease, clearExec func()) bool

Hold atomically registers combined ownership for workID exactly once when no terminal tombstone exists. Prefer BeginAdoption/Publish for accept/replay paths that race with MarkTerminal. Returns false when rejected; the caller must unwind the candidate and must not invoke clearExec against a winner.

func (*GenerationPinTracker) Len 

func (t *GenerationPinTracker) Len() int

Len reports outstanding held ownership entries (test/diagnostics).

func (*GenerationPinTracker) MarkTerminal 

func (t *GenerationPinTracker) MarkTerminal(workID string)

MarkTerminal records a definitive terminal transition before removing ownership. In-flight tokens cannot Publish afterward. Ownership is released outside the mutex with panic isolation. Tombstones without remaining reservations are dropped immediately.

func (*GenerationPinTracker) OwnershipSafe 

func (t *GenerationPinTracker) OwnershipSafe(workID string) bool

OwnershipSafe reports whether an existing tracker entry already proves ownership is handled (published ownership or terminal tombstone), so a replay candidate may release without publishing a replacement hold.

func (*GenerationPinTracker) Release 

func (t *GenerationPinTracker) Release(workID string)

Release drops combined ownership without marking terminal (nonterminal rollback / abort when durable append definitively did not happen). Does not create a terminal tombstone.

type IntentLookup 

type IntentLookup interface {
	LookupIntent(ctx context.Context, workID string) (rec terminalwork.WorkRecord, found bool, err error)
}

IntentLookup reconciles AppendIntent outcomes with unambiguous (record, found, error) semantics. Production stores implement this; absence of a definitive lookup is treated as ambiguous ownership.

type IntentService 

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

IntentService accepts durable settle/release intents with privacy-safe rows (requirements 8.3, 8.7–8.9, 12.8; design D9, D14).

func NewIntentService 

func NewIntentService(store IntentStore, cfg IntentServiceConfig) *IntentService

NewIntentService returns an intent accepter backed by store.

func (*IntentService) AcceptLeaseSetRelease 

func (s *IntentService) AcceptLeaseSetRelease(ctx context.Context, in LeaseSetReleaseInput) error

AcceptLeaseSetRelease appends and promotes a release-lease-set intent.

func (*IntentService) AcceptReleaseFailure 

func (s *IntentService) AcceptReleaseFailure(ctx context.Context, in ReleaseFailureInput) error

AcceptReleaseFailure appends and promotes a release-request-provider intent.

func (*IntentService) AcceptSettleFailure 

func (s *IntentService) AcceptSettleFailure(ctx context.Context, in SettleFailureInput) error

AcceptSettleFailure appends and promotes a settle-request-provider intent. Raw cause text is never persisted (design D14). WorkID/SourceKey are hash-based and include AttemptID so repeated request actions do not collide. Payload handles are sorted identically to the identity material for SameIntentReplay.

func (*IntentService) SetExecutablePending 

func (s *IntentService) SetExecutablePending(b ExecutablePendingBinder)

SetExecutablePending wires WorkID-keyed executable pending ownership after composition (e.g. when SnapshotPublisher becomes available).

type IntentServiceConfig 

type IntentServiceConfig struct {
	Clock func() time.Time
	// Pins retains runtime-generation pins for accepted durable work (task 3.6).
	// Nil disables pinning (legacy/test paths).
	Pins *GenerationPinTracker
	// ExecutablePending binds WorkID-keyed executable-generation pending ownership.
	ExecutablePending ExecutablePendingBinder
	// AmbiguousHandoff receives candidate ownership when append outcome is
	// ambiguous (subtask B). Nil releases the candidate and returns
	// ErrAmbiguousAppendReconcilerNotConfigured.
	AmbiguousHandoff AmbiguousAppendHandoff
}

IntentServiceConfig configures IntentService clocks and optional runtime generation pin tracking for durable intents that outlive the HTTP lease.

type IntentStore 

type IntentStore interface {
	AppendIntent(ctx context.Context, rec terminalwork.WorkRecord) error
	PromotePending(ctx context.Context, cmd terminalwork.PromotePendingCommand) error
}

IntentStore persists durable terminal-work intents before promotion.

type LeaseSetEffectConfig 

type LeaseSetEffectConfig struct {
	ProviderID string
	Provider   authority.ConcurrencyProvider
	Version    string
}

LeaseSetEffectConfig wires concurrency release for terminal lease-set work.

type LeaseSetEffectProvider 

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

LeaseSetEffectProvider completes WorkKindReleaseLeaseSet via ReleaseLease(SetID).

func NewLeaseSetEffectProvider 

func NewLeaseSetEffectProvider(cfg LeaseSetEffectConfig) (*LeaseSetEffectProvider, error)

NewLeaseSetEffectProvider validates and returns a lease-set effect adapter.

func (*LeaseSetEffectProvider) Invoke 

func (p *LeaseSetEffectProvider) Invoke(ctx context.Context, rec terminalwork.WorkRecord, _ string) error

Invoke releases the lease set; capacity is not freed until the store confirms.

func (*LeaseSetEffectProvider) ProviderID 

func (p *LeaseSetEffectProvider) ProviderID() string

func (*LeaseSetEffectProvider) SupportedKinds 

func (p *LeaseSetEffectProvider) SupportedKinds() []sdk.WorkKind

func (*LeaseSetEffectProvider) Version 

func (p *LeaseSetEffectProvider) Version() string

type LeaseSetReleaseInput 

type LeaseSetReleaseInput struct {
	RequestID  string
	AttemptID  string
	TraceID    string
	LeaseSetID string
	Reason     string
	Versions   terminalwork.BoundVersions
}

LeaseSetReleaseInput describes durable lease-set release/rollback work.

type MetricsConfig 

type MetricsConfig struct {
	Clock func() time.Time
}

MetricsConfig configures MetricsObserver clocks.

type MetricsObserver 

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

MetricsObserver computes backlog and oldest-age from a query store.

func NewMetricsObserver 

func NewMetricsObserver(store QueryStore, cfg MetricsConfig) *MetricsObserver

NewMetricsObserver returns a metrics snapshotter.

func (*MetricsObserver) Snapshot 

func (o *MetricsObserver) Snapshot(ctx context.Context) (MetricsSnapshot, error)

Snapshot returns counts and oldest outstanding age. Query failures and cursor faults are returned as errors; callers must not treat zero counts as ready.

type MetricsSnapshot 

type MetricsSnapshot struct {
	Backlog     int
	OldestAge   time.Duration
	Pending     int
	Retrying    int
	Quarantined int
	Completed   int
	Claimed     int
	Intent      int
}

MetricsSnapshot is a bounded backlog view for readiness/metrics exporters.

type OwnershipBinder 

type OwnershipBinder func() (release func(), created bool)

OwnershipBinder creates optional executable-generation pending ownership for the eligible tracker winner. It runs while GenerationPinTracker.mu is held and must not re-enter the tracker or block on external I/O. Production binders are in-memory / snapshotgen only (task 3.6).

created=true means this call established a hold and release must release it. created=false means an idempotent no-op; release must be nil / ignored.

type PinRelease 

type PinRelease interface {
	Release()
}

PinRelease releases a retained runtime-generation pin exactly once.

type ProcessMetrics 

type ProcessMetrics interface {
	ObserveTransition(state, kind, providerID string)
	RefreshAfterBatch(ctx context.Context)
}

type Processor 

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

Processor claims due work, invokes providers once per claim, and completes/retries/quarantines.

func NewProcessor 

func NewProcessor(store WorkStore, registry *Registry, cfg Config) (*Processor, error)

NewProcessor validates config and returns a ready processor.

func (*Processor) AddOnTerminalDone 

func (p *Processor) AddOnTerminalDone(fn func(rec terminalwork.WorkRecord))

AddOnTerminalDone registers an additional terminal-done subscriber. Safe under concurrent registration and invocation; does not overwrite existing callbacks (task 3.6). Prefer composing OnTerminalDone before Start.

func (*Processor) ProcessDue 

func (p *Processor) ProcessDue(ctx context.Context) error

func (*Processor) Readiness 

func (p *Processor) Readiness() Readiness

Readiness reports processor running state and unresolved provider IDs.

func (*Processor) Run 

func (p *Processor) Run(ctx context.Context) error

Run processes ticks until ctx is cancelled. Without TickC/TickInterval, runs ProcessDue once.

func (*Processor) Running 

func (p *Processor) Running() bool

Running reports whether Start owns an active Run goroutine.

func (*Processor) SetOnTerminalDone 

func (p *Processor) SetOnTerminalDone(fn func(rec terminalwork.WorkRecord))

SetOnTerminalDone appends a terminal-done callback (compat). Prefer AddOnTerminalDone or immutable Config.OnTerminalDone before Start.

func (*Processor) Shutdown 

func (p *Processor) Shutdown(ctx context.Context) error

Shutdown cancels Run and waits for the owned goroutine. On wait timeout, cancel remains in effect; started resets when Run exits.

func (*Processor) Start 

func (p *Processor) Start(parent context.Context) error

Start owns one Run goroutine until Shutdown. Waits out a prior run that is still exiting.

func (*Processor) UnresolvedProviderIDs 

func (p *Processor) UnresolvedProviderIDs() []string

UnresolvedProviderIDs returns provider IDs that failed resolution (pending drain).

type QueryClass 

type QueryClass string

QueryClass distinguishes supported operator terminal-work query shapes. 

const (
	QueryClassPendingTerminalWork QueryClass = "pending_terminal_work"
	QueryClassFinancialProjection QueryClass = "financial_projection"
)

type QueryService 

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

QueryService exposes bounded, privacy-safe terminal-work queries.

func NewQueryService 

func NewQueryService(store QueryStore) *QueryService

NewQueryService returns an operator query service.

func (*QueryService) List 

func (s *QueryService) List(ctx context.Context, q WorkQuery) (WorkPage, error)

List returns operator-safe rows or a bounded query rejection.

type QueryStore 

type QueryStore interface {
	List(ctx context.Context, q terminalwork.ListQuery) (terminalwork.ListPage, error)
}

QueryStore lists persisted terminal-work rows.

type Readiness 

type Readiness struct {
	Running               bool
	UnresolvedProviderIDs []string
}

Readiness is a content-safe processor status snapshot for composition/readiness (task 4.5).

type RecoveryStore 

type RecoveryStore interface {
	WorkStore
	IntentStore
	QueryStore
}

RecoveryStore is the composition-root terminal-work backing used for processor claims, durable intent accept, and bounded operator queries (task 4.5).

type Registry 

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

Registry resolves stable provider IDs to effect providers (design D4, D9).

func NewRegistry 

func NewRegistry() *Registry

NewRegistry returns an empty provider registry.

func (*Registry) Has 

func (r *Registry) Has(providerID string) bool

Has reports whether providerID is registered.

func (*Registry) ProviderIDs 

func (r *Registry) ProviderIDs() []string

ProviderIDs returns registered IDs (unsorted).

func (*Registry) Register 

func (r *Registry) Register(p EffectProvider) error

Register validates and indexes a provider. Duplicate IDs and typed nils fail.

func (*Registry) Resolve 

func (r *Registry) Resolve(providerID string, kind sdk.WorkKind) (EffectProvider, error)

Resolve returns the provider for id when it supports kind.

type ReleaseFailureInput 

type ReleaseFailureInput struct {
	RequestID  string
	AttemptID  string
	TraceID    string
	ProviderID string
	Handle     string
	Versions   terminalwork.BoundVersions
}

ReleaseFailureInput describes one failed provider release for durable recovery.

type SettleFailureInput 

type SettleFailureInput struct {
	RequestID  string
	AttemptID  string
	TraceID    string
	ProviderID string
	Handles    []string
	Versions   terminalwork.BoundVersions
}

SettleFailureInput describes one failed provider settle for durable recovery.

type TerminalProviderView 

type TerminalProviderView interface {
	Resolve(providerID string, kind sdk.WorkKind) (EffectProvider, error)
}

TerminalProviderView is an immutable generation-owned terminal-effect provider resolver. It never exposes mutable registries, Built, or config.

type Ticker 

type Ticker interface {
	C() <-chan time.Time
	Stop()
}

Ticker is a stoppable pulse source used for tick/renew loops.

type WorkPage 

type WorkPage struct {
	Rows   []WorkRow
	Cursor string
}

WorkPage is one bounded page of operator-safe rows.

type WorkQuery 

type WorkQuery struct {
	WorkID     string
	RequestID  string
	AttemptID  string
	ProviderID string
	Kind       sdk.WorkKind
	State      sdk.WorkState
	Class      QueryClass
	Limit      int
	Cursor     string
}

WorkQuery is a bounded operator filter for terminal-work rows.

type WorkRow 

type WorkRow struct {
	WorkID     string
	SourceKey  string
	Kind       sdk.WorkKind
	State      sdk.WorkState
	ProviderID string
	ErrorCode  string
	Payload    []byte
}

WorkRow is an operator-safe terminal-work projection (no raw payload/content).

type WorkStore 

type WorkStore interface {
	ClaimDue(ctx context.Context, cmd terminalwork.ClaimDueCommand) ([]terminalwork.WorkRecord, error)
	RenewClaim(ctx context.Context, cmd terminalwork.RenewClaimCommand) error
	Complete(ctx context.Context, cmd terminalwork.CompleteCommand) error
	ScheduleRetry(ctx context.Context, cmd terminalwork.ScheduleRetryCommand) error
	Quarantine(ctx context.Context, cmd terminalwork.QuarantineCommand) error
}

WorkStore is the consumer-owned persistence port for processor mutations. Adapters (memory/SQLite/PostgreSQL) implement these workstore commands.

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.