persistence

package
v19.2.0 Latest Latest
Warning

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

 Go to latest Published: Jul 15, 2026 License: MIT Imports: 22 Imported by: 0

Documentation

Index

Constants

View Source 
const ReservedIdentityKeyPrefix = "__autocore_internal:"

ReservedIdentityKeyPrefix is the prefix for autocore internal identity keys.

Variables

View Source 
var (
	CancellationReason_name = map[int32]string{
		0: "CANCELLATION_REASON_UNSPECIFIED",
		1: "CANCELLATION_REASON_EXPLICIT",
		2: "CANCELLATION_REASON_PARENT_TERMINATED",
		3: "CANCELLATION_REASON_SCHEDULE_TO_COMPLETE_TIMEOUT",
	}
	CancellationReason_value = map[string]int32{
		"CANCELLATION_REASON_UNSPECIFIED":                  0,
		"CANCELLATION_REASON_EXPLICIT":                     1,
		"CANCELLATION_REASON_PARENT_TERMINATED":            2,
		"CANCELLATION_REASON_SCHEDULE_TO_COMPLETE_TIMEOUT": 3,
	}
)

Enum value maps for CancellationReason.

View Source 
var (
	ErrFenced           = errors.New("shard ownership lost: fenced out")
	ErrRevisionConflict = errors.New("workflow execution revision conflict: concurrent modification")
)
View Source 
var ErrPayloadDecode = errors.New("payload decode failed")

ErrPayloadDecode wraps a failure to proto-decode a persisted payload (a bytea column previously written by autocore). It signals data corruption or schema skew rather than a transient error -- decoding the same bytes always fails -- so callers detect it via errors.Is to classify it as a non-retryable internal error and report it.

View Source 
var ErrWorkflowNotFound = errors.New("workflow execution not found")
View Source 
var File_internal_autocore_persistence_payloads_proto protoreflect.FileDescriptor

Functions

func AdvisoryUnlock added in v19.2.0 

func AdvisoryUnlock(ctx context.Context, db DB, name string) error

AdvisoryUnlock releases the session-level advisory lock keyed by name. db must be the same connection that took the lock.

func AppendHistoryEvents 

func AppendHistoryEvents(ctx context.Context, tx pgx.Tx, events []HistoryEventInsert) error

func BatchRecoverRunningActivityTasks 

func BatchRecoverRunningActivityTasks(ctx context.Context, tx pgx.Tx, shardID ShardID, limit int64) (int64, error)

BatchRecoverRunningActivityTasks recovers up to limit orphaned running activity tasks for a shard by re-enqueuing them to pending. Called on shard claim to clean up after the previous owner.

func BatchRecoverRunningWorkflowTasks 

func BatchRecoverRunningWorkflowTasks(ctx context.Context, tx pgx.Tx, shardID ShardID, limit int64) (int64, error)

BatchRecoverRunningWorkflowTasks recovers up to limit orphaned running workflow tasks for a shard. Tasks without a pending counterpart are re-enqueued to pending; tasks that already have a pending counterpart are deleted as redundant. Called on shard claim to clean up after the previous owner.

func BulkInsertChildWorkflowCancelSignals 

func BulkInsertChildWorkflowCancelSignals(ctx context.Context, tx pgx.Tx, shardID ShardID, parentID WorkflowID) (int64, error)

BulkInsertChildWorkflowCancelSignals inserts a SignalTypeCancelWorkflow row on the inbox of every pending non-detached child of parentID on this shard. The scanner promotes them to WorkflowCancellationRequested events on each child's history. Idempotent across retries via each child's (workflow_id, identity_key) row in signal_identity. Intended to run inside the parent's terminal fenced shard transaction. Returns the number of inbox rows inserted (new cancels).

func CreateActivityTasks 

func CreateActivityTasks(ctx context.Context, tx pgx.Tx, rows []*ActivityTaskInsert) error

func CreateNewChildWorkflowExecution 

func CreateNewChildWorkflowExecution(ctx context.Context, tx pgx.Tx, row *WorkflowExecutionInsert) (time.Time, bool, error)

CreateNewChildWorkflowExecution inserts a child workflow_execution row, linking it to its parent via parent_workflow_id and the detached_from_parent flag. It uses ON CONFLICT (shard_id, workflow_id) DO NOTHING for idempotency: returns (createdAt, true, nil) on insert and (zero, false, nil) when a row for the same (shard_id, workflow_id) already exists. Child at-most-once scheduling is guaranteed by the parent's fenced persist tx and replay short-circuiting; this conflict path is unreachable in correct operation and a false return signals a determinism bug to the caller.

func CreateScheduledTask 

func CreateScheduledTask(ctx context.Context, tx pgx.Tx, row *ScheduledTaskInsert, fireAfter time.Duration) (time.Time, time.Duration, error)

CreateScheduledTask inserts a single scheduled task. Thin wrapper over CreateScheduledTasks for the single-row retry/abort paths.

func CreateShardMappings 

func CreateShardMappings(ctx context.Context, db DB, dbID DBID, shardIDs []ShardID) error

func CreateWorkflowDatabase 

func CreateWorkflowDatabase(ctx context.Context, db DB, row *WorkflowDatabaseRow) error

func CreateWorkflowExecution 

func CreateWorkflowExecution(ctx context.Context, tx pgx.Tx, row *WorkflowExecutionInsert) (time.Time, error)

CreateWorkflowExecution inserts a top-level workflow_execution row. Use CreateNewChildWorkflowExecution for awaitable or detached child workflows.

func CreateWorkflowTask 

func CreateWorkflowTask(ctx context.Context, tx pgx.Tx, row *WorkflowTaskInsert) (bool, error)

func DeleteActivityTask 

func DeleteActivityTask(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, activityTaskID ActivityTaskID) error

func DeleteFiredScheduledTask 

func DeleteFiredScheduledTask(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, id ScheduledTaskID) (time.Duration, error)

DeleteFiredScheduledTask deletes a scheduled_task that has just fired and returns the firing latency: the duration between its intended fire_at and the moment it was deleted, measured on the database clock (clock_timestamp() - fire_at). Both operands are DB-clock instants, so the result is free of app/DB skew, and the read rides on the delete the fire path already performs (no extra round-trip). This is the firing latency for the schedule-to-start metric; the drop paths use DeleteScheduledTask, which records no firing latency.

func DeleteScheduledTask 

func DeleteScheduledTask(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, id ScheduledTaskID) error

func DeleteScheduledTasksForWorkflow added in v19.2.0 

func DeleteScheduledTasksForWorkflow(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID) error

DeleteScheduledTasksForWorkflow removes every scheduled_task for a workflow. Used on terminal transitions to drop any timers left pending by selects/sleeps/retries. Uses the (shard_id, workflow_id, ...) primary-key prefix as a range delete.

func DeleteWorkflowTask 

func DeleteWorkflowTask(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, workflowTaskID WorkflowTaskID) error

func DetachBucket added in v19.2.0 

func DetachBucket(ctx context.Context, db DB, b Bucket) ([]string, error)

DetachBucket detaches every still-attached partition of the bucket, execution table first, and returns the partitions it detached this call. It is idempotent: a partition already detached by an interrupted prior cycle is skipped, so a half-finished retirement completes on a later sweep. Each detach runs outside a transaction (see DetachPartitionConcurrently).

func DetachPartitionConcurrently added in v19.2.0 

func DetachPartitionConcurrently(ctx context.Context, db DB, table, partition string) error

DetachPartitionConcurrently detaches partition from table without blocking reads or writes on the rest of the table. It must run outside a transaction block (CONCURRENTLY forbids one) on the caller's advisory-locked connection. The detached table stays on disk as a standalone table until DropDetachedPartitions removes it; an interrupted detach leaves it half-done for FinalizePendingDetaches.

func DropDetachedPartitions added in v19.2.0 

func DropDetachedPartitions(ctx context.Context, db DB) ([]string, error)

DropDetachedPartitions drops every standalone table left behind by a completed detach and returns their names. Runs on the advisory-locked connection.

func EnsurePartitions added in v19.2.0 

func EnsurePartitions(ctx context.Context, db DB, from, to time.Time, granularity time.Duration) error

EnsurePartitions creates any missing partitions covering [from, to] on the three retained tables, one per granularity-sized bucket. Buckets are anchored at the Unix epoch (bucket k spans [k*granularity, (k+1)*granularity)), so they are stable across calls and processes and abut exactly. db must hold partition-DDL privilege (the table owner / migration role).

It is idempotent and safe under concurrent callers: CREATE ... IF NOT EXISTS is not atomic, so when two instances race to create the same partition the loser gets duplicate_table (42P07), which is treated as success.

func FinalizePendingDetaches added in v19.2.0 

func FinalizePendingDetaches(ctx context.Context, db DB) ([]string, error)

FinalizePendingDetaches completes every partition left half-detached by an interrupted DETACH ... CONCURRENTLY and returns their names. Runs on the advisory-locked connection.

func InsertSignal 

func InsertSignal(ctx context.Context, db DB, row *SignalRow) error

InsertSignal writes a signal into signal_inbox and records its identity in signal_identity, atomically in one statement, for the workflow identified by (ShardID, WorkflowID). A re-send with an identity_key already present in signal_identity conflicts and inserts no inbox row, so the signal is not re-delivered (at-most-once), even after the original inbox row was consumed. Returns ErrWorkflowNotFound if no such workflow execution exists.

func IsBucketDrainable added in v19.2.0 

func IsBucketDrainable(ctx context.Context, db DB, b Bucket, retentionWindow time.Duration) (bool, error)

IsBucketDrainable reports whether no workflow in the bucket has been updated within retentionWindow, measured on the DB clock. A drainable bucket can be retired: all three of its partitions can be detached and dropped. The activity signal is workflow_execution.updated_at, the only mutable column among the three tables.

A bucket whose workflow_execution partition has already been dropped is mid-retirement -- it passed the gate when it was detached -- so it reports drainable, letting a sweep finish detaching the remaining partitions.

func LockWorkflowExecution 

func LockWorkflowExecution(ctx context.Context, tx pgx.Tx, shardID ShardID, id WorkflowID) error

LockWorkflowExecution takes the per-workflow FOR UPDATE lock without reading any columns. Use it when a transaction only needs to serialize behind other writers of the same workflow. Returns pgx.ErrNoRows if the workflow execution does not exist.

func PollDeadlineReached added in v19.2.0 

func PollDeadlineReached(ctx context.Context, tx pgx.Tx, deadline time.Time) (bool, error)

PollDeadlineReached reports whether the poll deadline has passed, measured on the DB clock (clock_timestamp()). The deadline originates from the workflow's Now() -- a history event timestamp, which is itself DB-clock -- so this never compares DB and application wall-clock times.

func RandomIdentityKey added in v19.2.0 

func RandomIdentityKey() string

func SecondsSince 

func SecondsSince(ctx context.Context, tx pgx.Tx, t time.Time) (float64, error)

SecondsSince returns the elapsed seconds, measured on the database clock, from t until clock_timestamp() evaluated now inside tx. Both t and clock_timestamp() are DB-clock instants, so the result is free of app/DB skew. Signal promotion uses it to read promote_complete after the create/dedup, where (unlike the scheduled_task fire paths) there is no row write to ride a RETURNING on.

func SetActivityTaskAbortRetrying 

func SetActivityTaskAbortRetrying(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, activityTaskID ActivityTaskID) error

SetActivityTaskAbortRetrying transitions a running activity task to retrying state during abort retry backoff and increments abort_retry_attempt. Does NOT bump attempt (reserved for app-level retries) and does NOT reset abort_retry_attempt. Caller is responsible for creating the matching activity-retry scheduled_task row within the same fenced transaction.

func SetActivityTaskPending 

func SetActivityTaskPending(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, activityTaskID ActivityTaskID) (bool, error)

SetActivityTaskPending resets a retrying activity task to pending after a retry backoff has elapsed. For an app retry (abort_retry_attempt == 0) it advances due_at to now, so the schedule-to-start metric on the next attempt excludes the user-requested backoff and starts where the activity_retry scheduled_task's firing latency ends. For an abort retry (abort_retry_attempt > 0) due_at is left as-is, so the engine-added abort backoff folds in.

func SetActivityTaskPollRetrying added in v19.2.0 

func SetActivityTaskPollRetrying(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, activityTaskID ActivityTaskID) error

SetActivityTaskPollRetrying transitions a running poll activity to retrying state during a not-ready re-poll. Polling is bounded by the poll deadline, not the failure RetryPolicy: a not-ready execution proves the activity works, so attempt resets to 1 and RetryPolicy.MaxAttempts bounds consecutive failures within a poll rather than failures across its lifetime. abort_retry_attempt resets to 0 so the re-pend advances due_at past the poll interval. Caller creates the matching activity-retry scheduled_task row in the same tx.

func SetActivityTaskRetrying 

func SetActivityTaskRetrying(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, activityTaskID ActivityTaskID) error

SetActivityTaskRetrying transitions a running activity task to retrying state during app-retry backoff, bumps attempt, and resets abort_retry_attempt to 0. The reset also marks the next re-pend as an app retry, so SetActivityTaskPending advances due_at past the user backoff.

func SetWorkflowTaskAbortRetrying 

func SetWorkflowTaskAbortRetrying(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, workflowTaskID WorkflowTaskID) (bool, error)

SetWorkflowTaskAbortRetrying transitions a running workflow task to retrying state during abort retry backoff and increments abort_retry_attempt. Returns false (no error) if a Pending counterpart already exists for the same workflow -- the caller should delete the Running row instead, since the Pending counterpart will drive replay. Returns false (no error) if the row is no longer in Running state (recovery already swept it).

Caller is responsible for creating the matching workflow-retry scheduled_task row within the same fenced transaction when this returns true.

func SetWorkflowTaskPending 

func SetWorkflowTaskPending(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, workflowTaskID WorkflowTaskID) (bool, error)

SetWorkflowTaskPending resets a retrying workflow task to pending after a retry backoff has elapsed.

func TryAdvisoryLock added in v19.2.0 

func TryAdvisoryLock(ctx context.Context, db DB, name string) (bool, error)

TryAdvisoryLock attempts, without blocking, to take the session-level advisory lock keyed by name and reports whether it was acquired. The lock is held on db's connection until AdvisoryUnlock or the session ends, so db must be a single pinned connection, never a pool.

func TryCreateWorkflowExecution 

func TryCreateWorkflowExecution(ctx context.Context, tx pgx.Tx, row *WorkflowExecutionInsert) (time.Time, bool, error)

TryCreateWorkflowExecution is like CreateWorkflowExecution but uses ON CONFLICT DO NOTHING for idempotency. Returns (createdAt, true, nil) on insert, (zero, false, nil) if the row already exists.

func UpdateWorkflowExecution 

func UpdateWorkflowExecution(ctx context.Context, tx pgx.Tx, shardID ShardID, id WorkflowID, expectedRevision int64, state WorkflowExecutionState, nextSequenceID SequenceID, signalEventCount int32) error

UpdateWorkflowExecution advances a workflow execution's mutable columns (state, next_sequence_id, signal_event_count) and increments the revision counter, checking that expectedRevision still matches. Callers pass the new absolute values; signalEventCount is normally the loaded row's value and is bumped by one only when promoting a signal event. The payload column is write-once at creation and is intentionally not rewritten here. Returns ErrRevisionConflict when the WHERE clause finds no rows.

Contract for callers: expectedRevision MUST come from a LockAndGetWorkflowExecution call in the same transaction (so that its FOR UPDATE lock blocks concurrent writers for the duration of tx). Under that contract, a second writer on the same row blocks at LockAndGetWorkflowExecution, re-reads the current committed revision after the prior writer commits, and its UPDATE then matches -- so ErrRevisionConflict is unreachable in practice. The revision check remains as a safety net: if a caller ever used a stale revision obtained outside the tx (or without the FOR UPDATE anchor), the UPDATE would surface the bug as ErrRevisionConflict rather than silently lose the prior writer's work.

func UpsertShardMappings 

func UpsertShardMappings(ctx context.Context, db DB, dbID DBID, shardIDs []ShardID) error

UpsertShardMappings does not update shard to DB mapping on conflict as this is not safe to do in production.

func UpsertWorkflowDatabase 

func UpsertWorkflowDatabase(ctx context.Context, db DB, row *WorkflowDatabaseRow) error

Types

type ActivityTaskID 

type ActivityTaskID uuid.UUID //nolint: recvcheck

ActivityTaskID identifies an activity task within a workflow.

func NewActivityTaskID 

func NewActivityTaskID() ActivityTaskID

func (ActivityTaskID) Bytes 

func (id ActivityTaskID) Bytes() []byte

Bytes returns the UUID as a byte slice, for use in protobuf fields.

func (*ActivityTaskID) ScanUUID 

func (id *ActivityTaskID) ScanUUID(v pgtype.UUID) error

func (ActivityTaskID) String 

func (id ActivityTaskID) String() string

func (ActivityTaskID) UUIDValue 

func (id ActivityTaskID) UUIDValue() (pgtype.UUID, error)

type ActivityTaskInsert 

type ActivityTaskInsert struct {
	ShardID           ShardID
	WorkflowID        WorkflowID
	ActivityTaskID    ActivityTaskID
	Payload           []byte
	State             ActivityTaskState
	WorkflowCreatedAt time.Time
}

ActivityTaskInsert holds the columns written when creating an activity_task.

type ActivityTaskPayload 

type ActivityTaskPayload struct {
	ActivityName             string                             `protobuf:"bytes,1,opt,name=activity_name,json=activityName" json:"activity_name,omitempty"`
	Input                    []byte                             `protobuf:"bytes,2,opt,name=input" json:"input,omitempty"`
	RetryPolicy              *RetryPolicy                       `protobuf:"bytes,3,opt,name=retry_policy,json=retryPolicy" json:"retry_policy,omitempty"`
	Observability            *ActivityTaskPayload_Observability `protobuf:"bytes,4,opt,name=observability" json:"observability,omitempty"`
	ClaimToCompleteTimeout   *durationpb.Duration               `protobuf:"bytes,5,opt,name=claim_to_complete_timeout,json=claimToCompleteTimeout" json:"claim_to_complete_timeout,omitempty"`
	ScheduledEventSequenceId int64                              `` /* 131-byte string literal not displayed */
	Poll                     *PollSpec                          `protobuf:"bytes,7,opt,name=poll" json:"poll,omitempty"`
	// contains filtered or unexported fields
}

func (*ActivityTaskPayload) Descriptor deprecated 

func (*ActivityTaskPayload) Descriptor() ([]byte, []int)

Deprecated: Use ActivityTaskPayload.ProtoReflect.Descriptor instead.

func (*ActivityTaskPayload) GetActivityName 

func (x *ActivityTaskPayload) GetActivityName() string

func (*ActivityTaskPayload) GetClaimToCompleteTimeout 

func (x *ActivityTaskPayload) GetClaimToCompleteTimeout() *durationpb.Duration

func (*ActivityTaskPayload) GetInput 

func (x *ActivityTaskPayload) GetInput() []byte

func (*ActivityTaskPayload) GetObservability 

func (x *ActivityTaskPayload) GetObservability() *ActivityTaskPayload_Observability

func (*ActivityTaskPayload) GetPoll added in v19.2.0 

func (x *ActivityTaskPayload) GetPoll() *PollSpec

func (*ActivityTaskPayload) GetRetryPolicy 

func (x *ActivityTaskPayload) GetRetryPolicy() *RetryPolicy

func (*ActivityTaskPayload) GetScheduledEventSequenceId 

func (x *ActivityTaskPayload) GetScheduledEventSequenceId() int64

func (*ActivityTaskPayload) ProtoMessage 

func (*ActivityTaskPayload) ProtoMessage()

func (*ActivityTaskPayload) ProtoReflect 

func (x *ActivityTaskPayload) ProtoReflect() protoreflect.Message

func (*ActivityTaskPayload) Reset 

func (x *ActivityTaskPayload) Reset()

func (*ActivityTaskPayload) ScanBytes 

func (x *ActivityTaskPayload) ScanBytes(src []byte) error

func (*ActivityTaskPayload) String 

func (x *ActivityTaskPayload) String() string

type ActivityTaskPayload_Observability 

type ActivityTaskPayload_Observability struct {
	Parent         *SpanContext `protobuf:"bytes,1,opt,name=parent" json:"parent,omitempty"`
	SchedulingSpan *SpanContext `protobuf:"bytes,2,opt,name=scheduling_span,json=schedulingSpan" json:"scheduling_span,omitempty"`
	// contains filtered or unexported fields
}

func (*ActivityTaskPayload_Observability) Descriptor deprecated 

func (*ActivityTaskPayload_Observability) Descriptor() ([]byte, []int)

Deprecated: Use ActivityTaskPayload_Observability.ProtoReflect.Descriptor instead.

func (*ActivityTaskPayload_Observability) GetParent 

func (x *ActivityTaskPayload_Observability) GetParent() *SpanContext

func (*ActivityTaskPayload_Observability) GetSchedulingSpan 

func (x *ActivityTaskPayload_Observability) GetSchedulingSpan() *SpanContext

func (*ActivityTaskPayload_Observability) ProtoMessage 

func (*ActivityTaskPayload_Observability) ProtoMessage()

func (*ActivityTaskPayload_Observability) ProtoReflect 

func (x *ActivityTaskPayload_Observability) ProtoReflect() protoreflect.Message

func (*ActivityTaskPayload_Observability) Reset 

func (x *ActivityTaskPayload_Observability) Reset()

func (*ActivityTaskPayload_Observability) String 

func (x *ActivityTaskPayload_Observability) String() string

type ActivityTaskRow 

type ActivityTaskRow struct {
	ShardID           ShardID
	WorkflowID        WorkflowID
	ActivityTaskID    ActivityTaskID
	Payload           *ActivityTaskPayload
	State             ActivityTaskState
	Attempt           int32
	AbortRetryAttempt int32
	ClaimedAt         *time.Time
	CreatedAt         time.Time
	DueAt             time.Time
	WorkflowCreatedAt time.Time
}

ActivityTaskRow is the read model scanned from activity_task. Its Payload is decoded into an ActivityTaskPayload during the scan.

func BulkClaimActivityTasks 

func BulkClaimActivityTasks(ctx context.Context, tx pgx.Tx, shardIDs []ShardID, perShardLimit, totalLimit int) ([]ActivityTaskRow, error)

type ActivityTaskState 

type ActivityTaskState int16

ActivityTaskState represents the lifecycle state of an activity task. Important: do not change constants! They are used in DB data. 

const (
	ActivityTaskStateUnknown  ActivityTaskState = 0
	ActivityTaskStatePending  ActivityTaskState = 1
	ActivityTaskStateRunning  ActivityTaskState = 2
	ActivityTaskStateRetrying ActivityTaskState = 3
)

func (ActivityTaskState) String 

func (s ActivityTaskState) String() string

type Bucket added in v19.2.0 

type Bucket struct {
	Suffix string
	Start  time.Time
}

Bucket identifies one partition bucket shared across the three retained tables. Suffix is the _p_<YYYYMMDD> bucket-start date; Start is that date as an instant.

func RetirableBuckets added in v19.2.0 

func RetirableBuckets(ctx context.Context, db DB, olderThan time.Time, granularity time.Duration) ([]Bucket, error)

RetirableBuckets returns the buckets whose entire era ended at or before olderThan -- every workflow they could hold was created more than the retention window ago -- making them candidates for the drainability check. The newest bucket never qualifies: the sweep extends the runway first, so a future bucket always sits beyond olderThan. Bounds come from partition names, never the live clock.

type CancellationReason added in v19.2.0 

type CancellationReason int32
const (
	CancellationReason_CANCELLATION_REASON_UNSPECIFIED                  CancellationReason = 0
	CancellationReason_CANCELLATION_REASON_EXPLICIT                     CancellationReason = 1
	CancellationReason_CANCELLATION_REASON_PARENT_TERMINATED            CancellationReason = 2
	CancellationReason_CANCELLATION_REASON_SCHEDULE_TO_COMPLETE_TIMEOUT CancellationReason = 3
)

func (CancellationReason) Descriptor added in v19.2.0 

func (CancellationReason) Descriptor() protoreflect.EnumDescriptor

func (CancellationReason) Enum added in v19.2.0 

func (x CancellationReason) Enum() *CancellationReason

func (CancellationReason) EnumDescriptor deprecated added in v19.2.0 

func (CancellationReason) EnumDescriptor() ([]byte, []int)

Deprecated: Use CancellationReason.Descriptor instead.

func (CancellationReason) Number added in v19.2.0 

func (x CancellationReason) Number() protoreflect.EnumNumber

func (CancellationReason) String added in v19.2.0 

func (x CancellationReason) String() string

func (CancellationReason) Type added in v19.2.0 

func (CancellationReason) Type() protoreflect.EnumType

type CentralStore 

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

CentralStore owns queries against the central coordination database that tracks workflow databases, shard mappings, and cross-DB identity dedup.

func NewCentralStore 

func NewCentralStore(pool *pgxpool.Pool) *CentralStore

func (*CentralStore) CreateDBWithShards 

func (s *CentralStore) CreateDBWithShards(ctx context.Context, dbRow *WorkflowDatabaseRow, shardIDs []ShardID) error

func (*CentralStore) DB 

func (s *CentralStore) DB() DB

func (*CentralStore) DeleteWorkflowIdentity 

func (s *CentralStore) DeleteWorkflowIdentity(ctx context.Context, namespaceID NamespaceID, identityKey string) error

DeleteWorkflowIdentity removes a workflow identity row. This is a best-effort cleanup operation used when workflow creation fails after the identity was pinned.

func (*CentralStore) ExistingShardIDs 

func (s *CentralStore) ExistingShardIDs(ctx context.Context, shardIDs []ShardID) ([]ShardID, error)

ExistingShardIDs returns the subset of shardIDs already present in the shard table, sorted by shard ID.

func (*CentralStore) GetWorkflowDatabaseWithStats 

func (s *CentralStore) GetWorkflowDatabaseWithStats(ctx context.Context, dbID DBID) (WorkflowDatabaseWithStats, error)

GetWorkflowDatabaseWithStats returns a single workflow database with its shard-allocation summary, fetched in one query. Returns pgx.ErrNoRows if no database has dbID.

func (*CentralStore) ListShardMappings 

func (s *CentralStore) ListShardMappings(ctx context.Context) ([]ShardMappingRow, error)

func (*CentralStore) ListWorkflowDatabases 

func (s *CentralStore) ListWorkflowDatabases(ctx context.Context) ([]WorkflowDatabaseRow, error)

func (*CentralStore) ListWorkflowDatabasesWithStats 

func (s *CentralStore) ListWorkflowDatabasesWithStats(ctx context.Context) ([]WorkflowDatabaseWithStats, error)

ListWorkflowDatabasesWithStats returns every workflow database with its shard-allocation summary, ordered by db_id, in one query.

func (*CentralStore) TryCreateWorkflowIdentity 

func (s *CentralStore) TryCreateWorkflowIdentity(
	ctx context.Context,
	namespaceID NamespaceID,
	identityKey string,
	workflowID WorkflowID,
	shardID ShardID,
) (WorkflowIdentityResult, error)

TryCreateWorkflowIdentity attempts to insert a new workflow identity row. If the (namespace_id, identity_key) pair already exists, the existing row's workflow_id and shard_id are returned. This ensures that retries after a partial failure route to the same shard/database.

func (*CentralStore) UpsertDBWithShards 

func (s *CentralStore) UpsertDBWithShards(ctx context.Context, dbRow *WorkflowDatabaseRow, shardIDs []ShardID) error

UpsertDBWithShards ensures workflow DB record and specified shard mappings exist. Warning: do not use in production code!

func (*CentralStore) UpsertNamespace 

func (s *CentralStore) UpsertNamespace(ctx context.Context, externalNamespaceID ExternalNamespaceID) (NamespaceID, error)

UpsertNamespace maps an externally-supplied namespace id to autocore's internal namespace surrogate, creating the namespace row on first use. It is idempotent: repeated calls for the same external id return the same internal id.

type ClaimedShard 

type ClaimedShard struct {
	ShardID ShardID
	FenceID FenceID
}

type DB 

type DB interface {
	Begin(ctx context.Context) (pgx.Tx, error)
	CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
	Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}

DB can be used where no transaction is required.

type DBID 

type DBID int16

type ExternalNamespaceID 

type ExternalNamespaceID int64

ExternalNamespaceID is the namespace id supplied by the caller when starting a workflow. The namespace table maps it to an autocore-internal NamespaceID.

type FenceID 

type FenceID int64

type HistoryEventDebugPayload added in v19.2.0 

type HistoryEventDebugPayload struct {

	// Types that are valid to be assigned to Event:
	//
	//	*HistoryEventDebugPayload_ActivityScheduled
	//	*HistoryEventDebugPayload_InlineActivityCompleted
	//	*HistoryEventDebugPayload_InlineActivityFailed
	Event isHistoryEventDebugPayload_Event `protobuf_oneof:"event"`
	// contains filtered or unexported fields
}

func (*HistoryEventDebugPayload) Descriptor deprecated added in v19.2.0 

func (*HistoryEventDebugPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventDebugPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventDebugPayload) GetActivityScheduled added in v19.2.0 

func (x *HistoryEventDebugPayload) GetActivityScheduled() *HistoryEventDebugPayload_ActivityScheduledPayload

func (*HistoryEventDebugPayload) GetEvent added in v19.2.0 

func (x *HistoryEventDebugPayload) GetEvent() isHistoryEventDebugPayload_Event

func (*HistoryEventDebugPayload) GetInlineActivityCompleted added in v19.2.0 

func (x *HistoryEventDebugPayload) GetInlineActivityCompleted() *HistoryEventDebugPayload_InlineActivityCompletedPayload

func (*HistoryEventDebugPayload) GetInlineActivityFailed added in v19.2.0 

func (x *HistoryEventDebugPayload) GetInlineActivityFailed() *HistoryEventDebugPayload_InlineActivityFailedPayload

func (*HistoryEventDebugPayload) ProtoMessage added in v19.2.0 

func (*HistoryEventDebugPayload) ProtoMessage()

func (*HistoryEventDebugPayload) ProtoReflect added in v19.2.0 

func (x *HistoryEventDebugPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventDebugPayload) Reset added in v19.2.0 

func (x *HistoryEventDebugPayload) Reset()

func (*HistoryEventDebugPayload) String added in v19.2.0 

func (x *HistoryEventDebugPayload) String() string

type HistoryEventDebugPayload_ActivityScheduled added in v19.2.0 

type HistoryEventDebugPayload_ActivityScheduled struct {
	ActivityScheduled *HistoryEventDebugPayload_ActivityScheduledPayload `protobuf:"bytes,1,opt,name=activity_scheduled,json=activityScheduled,oneof"`
}

type HistoryEventDebugPayload_ActivityScheduledPayload added in v19.2.0 

type HistoryEventDebugPayload_ActivityScheduledPayload struct {
	Input           []byte `protobuf:"bytes,1,opt,name=input" json:"input,omitempty"`
	PredicateConfig []byte `protobuf:"bytes,2,opt,name=predicate_config,json=predicateConfig" json:"predicate_config,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventDebugPayload_ActivityScheduledPayload) Descriptor deprecated added in v19.2.0 

func (*HistoryEventDebugPayload_ActivityScheduledPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventDebugPayload_ActivityScheduledPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventDebugPayload_ActivityScheduledPayload) GetInput added in v19.2.0 

func (x *HistoryEventDebugPayload_ActivityScheduledPayload) GetInput() []byte

func (*HistoryEventDebugPayload_ActivityScheduledPayload) GetPredicateConfig added in v19.2.0 

func (x *HistoryEventDebugPayload_ActivityScheduledPayload) GetPredicateConfig() []byte

func (*HistoryEventDebugPayload_ActivityScheduledPayload) ProtoMessage added in v19.2.0 

func (*HistoryEventDebugPayload_ActivityScheduledPayload) ProtoMessage()

func (*HistoryEventDebugPayload_ActivityScheduledPayload) ProtoReflect added in v19.2.0 

func (x *HistoryEventDebugPayload_ActivityScheduledPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventDebugPayload_ActivityScheduledPayload) Reset added in v19.2.0 

func (x *HistoryEventDebugPayload_ActivityScheduledPayload) Reset()

func (*HistoryEventDebugPayload_ActivityScheduledPayload) String added in v19.2.0 

func (x *HistoryEventDebugPayload_ActivityScheduledPayload) String() string

type HistoryEventDebugPayload_InlineActivityCompleted added in v19.2.0 

type HistoryEventDebugPayload_InlineActivityCompleted struct {
	InlineActivityCompleted *HistoryEventDebugPayload_InlineActivityCompletedPayload `protobuf:"bytes,2,opt,name=inline_activity_completed,json=inlineActivityCompleted,oneof"`
}

type HistoryEventDebugPayload_InlineActivityCompletedPayload added in v19.2.0 

type HistoryEventDebugPayload_InlineActivityCompletedPayload struct {
	Input []byte `protobuf:"bytes,1,opt,name=input" json:"input,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) Descriptor deprecated added in v19.2.0 

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventDebugPayload_InlineActivityCompletedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) GetInput added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityCompletedPayload) GetInput() []byte

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) ProtoMessage added in v19.2.0 

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) ProtoMessage()

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) ProtoReflect added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityCompletedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) Reset added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityCompletedPayload) Reset()

func (*HistoryEventDebugPayload_InlineActivityCompletedPayload) String added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityCompletedPayload) String() string

type HistoryEventDebugPayload_InlineActivityFailed added in v19.2.0 

type HistoryEventDebugPayload_InlineActivityFailed struct {
	InlineActivityFailed *HistoryEventDebugPayload_InlineActivityFailedPayload `protobuf:"bytes,3,opt,name=inline_activity_failed,json=inlineActivityFailed,oneof"`
}

type HistoryEventDebugPayload_InlineActivityFailedPayload added in v19.2.0 

type HistoryEventDebugPayload_InlineActivityFailedPayload struct {
	Input []byte `protobuf:"bytes,1,opt,name=input" json:"input,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) Descriptor deprecated added in v19.2.0 

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventDebugPayload_InlineActivityFailedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) GetInput added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityFailedPayload) GetInput() []byte

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) ProtoMessage added in v19.2.0 

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) ProtoMessage()

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) ProtoReflect added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityFailedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) Reset added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityFailedPayload) Reset()

func (*HistoryEventDebugPayload_InlineActivityFailedPayload) String added in v19.2.0 

func (x *HistoryEventDebugPayload_InlineActivityFailedPayload) String() string

type HistoryEventInsert 

type HistoryEventInsert struct {
	ShardID      ShardID
	WorkflowID   WorkflowID
	SequenceID   SequenceID
	Payload      []byte
	DebugPayload []byte
}

HistoryEventInsert holds the columns written when appending a history_event. DebugPayload is the optional marshaled HistoryEventDebugPayload; nil for events that carry no replay-irrelevant debug data, stored as SQL NULL.

type HistoryEventPayload 

type HistoryEventPayload struct {

	// Types that are valid to be assigned to Event:
	//
	//	*HistoryEventPayload_WorkflowCreated
	//	*HistoryEventPayload_WorkflowCompleted
	//	*HistoryEventPayload_WorkflowFailed
	//	*HistoryEventPayload_ActivityScheduled
	//	*HistoryEventPayload_ActivityCompleted
	//	*HistoryEventPayload_ActivityFailed
	//	*HistoryEventPayload_TimerStarted
	//	*HistoryEventPayload_TimerFired
	//	*HistoryEventPayload_SignalReceived
	//	*HistoryEventPayload_WorkflowCancellationRequested
	//	*HistoryEventPayload_WorkflowCanceled
	//	*HistoryEventPayload_ChildWorkflowScheduled
	//	*HistoryEventPayload_ChildWorkflowCompleted
	//	*HistoryEventPayload_ChildWorkflowFailed
	//	*HistoryEventPayload_ChildWorkflowCanceled
	//	*HistoryEventPayload_InlineActivityCompleted
	//	*HistoryEventPayload_InlineActivityFailed
	//	*HistoryEventPayload_WorkflowTimedOut
	Event isHistoryEventPayload_Event `protobuf_oneof:"event"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload) Descriptor deprecated 

func (*HistoryEventPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload) GetActivityCompleted 

func (x *HistoryEventPayload) GetActivityCompleted() *HistoryEventPayload_ActivityCompletedPayload

func (*HistoryEventPayload) GetActivityFailed 

func (x *HistoryEventPayload) GetActivityFailed() *HistoryEventPayload_ActivityFailedPayload

func (*HistoryEventPayload) GetActivityScheduled 

func (x *HistoryEventPayload) GetActivityScheduled() *HistoryEventPayload_ActivityScheduledPayload

func (*HistoryEventPayload) GetChildWorkflowCanceled 

func (x *HistoryEventPayload) GetChildWorkflowCanceled() *HistoryEventPayload_ChildWorkflowCanceledPayload

func (*HistoryEventPayload) GetChildWorkflowCompleted 

func (x *HistoryEventPayload) GetChildWorkflowCompleted() *HistoryEventPayload_ChildWorkflowCompletedPayload

func (*HistoryEventPayload) GetChildWorkflowFailed 

func (x *HistoryEventPayload) GetChildWorkflowFailed() *HistoryEventPayload_ChildWorkflowFailedPayload

func (*HistoryEventPayload) GetChildWorkflowScheduled 

func (x *HistoryEventPayload) GetChildWorkflowScheduled() *HistoryEventPayload_ChildWorkflowScheduledPayload

func (*HistoryEventPayload) GetEvent 

func (x *HistoryEventPayload) GetEvent() isHistoryEventPayload_Event

func (*HistoryEventPayload) GetInlineActivityCompleted 

func (x *HistoryEventPayload) GetInlineActivityCompleted() *HistoryEventPayload_InlineActivityCompletedPayload

func (*HistoryEventPayload) GetInlineActivityFailed 

func (x *HistoryEventPayload) GetInlineActivityFailed() *HistoryEventPayload_InlineActivityFailedPayload

func (*HistoryEventPayload) GetSignalReceived 

func (x *HistoryEventPayload) GetSignalReceived() *HistoryEventPayload_SignalReceivedPayload

func (*HistoryEventPayload) GetTimerFired 

func (x *HistoryEventPayload) GetTimerFired() *HistoryEventPayload_TimerFiredPayload

func (*HistoryEventPayload) GetTimerStarted 

func (x *HistoryEventPayload) GetTimerStarted() *HistoryEventPayload_TimerStartedPayload

func (*HistoryEventPayload) GetWorkflowCanceled 

func (x *HistoryEventPayload) GetWorkflowCanceled() *HistoryEventPayload_WorkflowCanceledPayload

func (*HistoryEventPayload) GetWorkflowCancellationRequested 

func (x *HistoryEventPayload) GetWorkflowCancellationRequested() *HistoryEventPayload_WorkflowCancellationRequestedPayload

func (*HistoryEventPayload) GetWorkflowCompleted 

func (x *HistoryEventPayload) GetWorkflowCompleted() *HistoryEventPayload_WorkflowCompletedPayload

func (*HistoryEventPayload) GetWorkflowCreated 

func (x *HistoryEventPayload) GetWorkflowCreated() *HistoryEventPayload_WorkflowCreatedPayload

func (*HistoryEventPayload) GetWorkflowFailed 

func (x *HistoryEventPayload) GetWorkflowFailed() *HistoryEventPayload_WorkflowFailedPayload

func (*HistoryEventPayload) GetWorkflowTimedOut added in v19.2.0 

func (x *HistoryEventPayload) GetWorkflowTimedOut() *HistoryEventPayload_WorkflowTimedOutPayload

func (*HistoryEventPayload) ProtoMessage 

func (*HistoryEventPayload) ProtoMessage()

func (*HistoryEventPayload) ProtoReflect 

func (x *HistoryEventPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload) Reset 

func (x *HistoryEventPayload) Reset()

func (*HistoryEventPayload) ScanBytes 

func (x *HistoryEventPayload) ScanBytes(src []byte) error

func (*HistoryEventPayload) String 

func (x *HistoryEventPayload) String() string

type HistoryEventPayload_ActivityCompleted 

type HistoryEventPayload_ActivityCompleted struct {
	ActivityCompleted *HistoryEventPayload_ActivityCompletedPayload `protobuf:"bytes,5,opt,name=activity_completed,json=activityCompleted,oneof"`
}

type HistoryEventPayload_ActivityCompletedPayload 

type HistoryEventPayload_ActivityCompletedPayload struct {
	TaskId                   []byte `protobuf:"bytes,1,opt,name=task_id,json=taskId" json:"task_id,omitempty"`
	Result                   []byte `protobuf:"bytes,2,opt,name=result" json:"result,omitempty"`
	Attempt                  int32  `protobuf:"varint,3,opt,name=attempt" json:"attempt,omitempty"`
	ScheduledEventSequenceId int64  `` /* 131-byte string literal not displayed */
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_ActivityCompletedPayload) Descriptor deprecated 

func (*HistoryEventPayload_ActivityCompletedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_ActivityCompletedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_ActivityCompletedPayload) GetAttempt 

func (x *HistoryEventPayload_ActivityCompletedPayload) GetAttempt() int32

func (*HistoryEventPayload_ActivityCompletedPayload) GetResult 

func (x *HistoryEventPayload_ActivityCompletedPayload) GetResult() []byte

func (*HistoryEventPayload_ActivityCompletedPayload) GetScheduledEventSequenceId 

func (x *HistoryEventPayload_ActivityCompletedPayload) GetScheduledEventSequenceId() int64

func (*HistoryEventPayload_ActivityCompletedPayload) GetTaskId 

func (x *HistoryEventPayload_ActivityCompletedPayload) GetTaskId() []byte

func (*HistoryEventPayload_ActivityCompletedPayload) ProtoMessage 

func (*HistoryEventPayload_ActivityCompletedPayload) ProtoMessage()

func (*HistoryEventPayload_ActivityCompletedPayload) ProtoReflect 

func (x *HistoryEventPayload_ActivityCompletedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_ActivityCompletedPayload) Reset 

func (x *HistoryEventPayload_ActivityCompletedPayload) Reset()

func (*HistoryEventPayload_ActivityCompletedPayload) String 

func (x *HistoryEventPayload_ActivityCompletedPayload) String() string

type HistoryEventPayload_ActivityFailed 

type HistoryEventPayload_ActivityFailed struct {
	ActivityFailed *HistoryEventPayload_ActivityFailedPayload `protobuf:"bytes,6,opt,name=activity_failed,json=activityFailed,oneof"`
}

type HistoryEventPayload_ActivityFailedPayload 

type HistoryEventPayload_ActivityFailedPayload struct {
	TaskId                   []byte `protobuf:"bytes,1,opt,name=task_id,json=taskId" json:"task_id,omitempty"`
	Error                    string `protobuf:"bytes,2,opt,name=error" json:"error,omitempty"`
	Attempt                  int32  `protobuf:"varint,3,opt,name=attempt" json:"attempt,omitempty"`
	ScheduledEventSequenceId int64  `` /* 131-byte string literal not displayed */
	SystemFailure            bool   `protobuf:"varint,5,opt,name=system_failure,json=systemFailure" json:"system_failure,omitempty"`
	PollExhausted            bool   `protobuf:"varint,6,opt,name=poll_exhausted,json=pollExhausted" json:"poll_exhausted,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_ActivityFailedPayload) Descriptor deprecated 

func (*HistoryEventPayload_ActivityFailedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_ActivityFailedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_ActivityFailedPayload) GetAttempt 

func (x *HistoryEventPayload_ActivityFailedPayload) GetAttempt() int32

func (*HistoryEventPayload_ActivityFailedPayload) GetError 

func (x *HistoryEventPayload_ActivityFailedPayload) GetError() string

func (*HistoryEventPayload_ActivityFailedPayload) GetPollExhausted added in v19.2.0 

func (x *HistoryEventPayload_ActivityFailedPayload) GetPollExhausted() bool

func (*HistoryEventPayload_ActivityFailedPayload) GetScheduledEventSequenceId 

func (x *HistoryEventPayload_ActivityFailedPayload) GetScheduledEventSequenceId() int64

func (*HistoryEventPayload_ActivityFailedPayload) GetSystemFailure added in v19.2.0 

func (x *HistoryEventPayload_ActivityFailedPayload) GetSystemFailure() bool

func (*HistoryEventPayload_ActivityFailedPayload) GetTaskId 

func (x *HistoryEventPayload_ActivityFailedPayload) GetTaskId() []byte

func (*HistoryEventPayload_ActivityFailedPayload) ProtoMessage 

func (*HistoryEventPayload_ActivityFailedPayload) ProtoMessage()

func (*HistoryEventPayload_ActivityFailedPayload) ProtoReflect 

func (x *HistoryEventPayload_ActivityFailedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_ActivityFailedPayload) Reset 

func (x *HistoryEventPayload_ActivityFailedPayload) Reset()

func (*HistoryEventPayload_ActivityFailedPayload) String 

func (x *HistoryEventPayload_ActivityFailedPayload) String() string

type HistoryEventPayload_ActivityScheduled 

type HistoryEventPayload_ActivityScheduled struct {
	ActivityScheduled *HistoryEventPayload_ActivityScheduledPayload `protobuf:"bytes,4,opt,name=activity_scheduled,json=activityScheduled,oneof"`
}

type HistoryEventPayload_ActivityScheduledPayload 

type HistoryEventPayload_ActivityScheduledPayload struct {
	TaskId                 []byte               `protobuf:"bytes,1,opt,name=task_id,json=taskId" json:"task_id,omitempty"`
	ActivityName           string               `protobuf:"bytes,2,opt,name=activity_name,json=activityName" json:"activity_name,omitempty"`
	RetryPolicy            *RetryPolicy         `protobuf:"bytes,3,opt,name=retry_policy,json=retryPolicy" json:"retry_policy,omitempty"`
	ReplayKey              int32                `protobuf:"varint,4,opt,name=replay_key,json=replayKey" json:"replay_key,omitempty"`
	ClaimToCompleteTimeout *durationpb.Duration `protobuf:"bytes,5,opt,name=claim_to_complete_timeout,json=claimToCompleteTimeout" json:"claim_to_complete_timeout,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_ActivityScheduledPayload) Descriptor deprecated 

func (*HistoryEventPayload_ActivityScheduledPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_ActivityScheduledPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_ActivityScheduledPayload) GetActivityName 

func (x *HistoryEventPayload_ActivityScheduledPayload) GetActivityName() string

func (*HistoryEventPayload_ActivityScheduledPayload) GetClaimToCompleteTimeout 

func (x *HistoryEventPayload_ActivityScheduledPayload) GetClaimToCompleteTimeout() *durationpb.Duration

func (*HistoryEventPayload_ActivityScheduledPayload) GetReplayKey 

func (x *HistoryEventPayload_ActivityScheduledPayload) GetReplayKey() int32

func (*HistoryEventPayload_ActivityScheduledPayload) GetRetryPolicy 

func (x *HistoryEventPayload_ActivityScheduledPayload) GetRetryPolicy() *RetryPolicy

func (*HistoryEventPayload_ActivityScheduledPayload) GetTaskId 

func (x *HistoryEventPayload_ActivityScheduledPayload) GetTaskId() []byte

func (*HistoryEventPayload_ActivityScheduledPayload) ProtoMessage 

func (*HistoryEventPayload_ActivityScheduledPayload) ProtoMessage()

func (*HistoryEventPayload_ActivityScheduledPayload) ProtoReflect 

func (x *HistoryEventPayload_ActivityScheduledPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_ActivityScheduledPayload) Reset 

func (x *HistoryEventPayload_ActivityScheduledPayload) Reset()

func (*HistoryEventPayload_ActivityScheduledPayload) String 

func (x *HistoryEventPayload_ActivityScheduledPayload) String() string

type HistoryEventPayload_ChildWorkflowCanceled 

type HistoryEventPayload_ChildWorkflowCanceled struct {
	ChildWorkflowCanceled *HistoryEventPayload_ChildWorkflowCanceledPayload `protobuf:"bytes,15,opt,name=child_workflow_canceled,json=childWorkflowCanceled,oneof"`
}

type HistoryEventPayload_ChildWorkflowCanceledPayload 

type HistoryEventPayload_ChildWorkflowCanceledPayload struct {
	ChildWorkflowId []byte `protobuf:"bytes,1,opt,name=child_workflow_id,json=childWorkflowId" json:"child_workflow_id,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) Descriptor deprecated 

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_ChildWorkflowCanceledPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) GetChildWorkflowId 

func (x *HistoryEventPayload_ChildWorkflowCanceledPayload) GetChildWorkflowId() []byte

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) ProtoMessage 

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) ProtoMessage()

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) ProtoReflect 

func (x *HistoryEventPayload_ChildWorkflowCanceledPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) Reset 

func (x *HistoryEventPayload_ChildWorkflowCanceledPayload) Reset()

func (*HistoryEventPayload_ChildWorkflowCanceledPayload) String 

func (x *HistoryEventPayload_ChildWorkflowCanceledPayload) String() string

type HistoryEventPayload_ChildWorkflowCompleted 

type HistoryEventPayload_ChildWorkflowCompleted struct {
	ChildWorkflowCompleted *HistoryEventPayload_ChildWorkflowCompletedPayload `protobuf:"bytes,13,opt,name=child_workflow_completed,json=childWorkflowCompleted,oneof"`
}

type HistoryEventPayload_ChildWorkflowCompletedPayload 

type HistoryEventPayload_ChildWorkflowCompletedPayload struct {
	ChildWorkflowId []byte `protobuf:"bytes,1,opt,name=child_workflow_id,json=childWorkflowId" json:"child_workflow_id,omitempty"`
	Result          []byte `protobuf:"bytes,2,opt,name=result" json:"result,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) Descriptor deprecated 

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_ChildWorkflowCompletedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) GetChildWorkflowId 

func (x *HistoryEventPayload_ChildWorkflowCompletedPayload) GetChildWorkflowId() []byte

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) GetResult 

func (x *HistoryEventPayload_ChildWorkflowCompletedPayload) GetResult() []byte

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) ProtoMessage 

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) ProtoMessage()

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) ProtoReflect 

func (x *HistoryEventPayload_ChildWorkflowCompletedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) Reset 

func (x *HistoryEventPayload_ChildWorkflowCompletedPayload) Reset()

func (*HistoryEventPayload_ChildWorkflowCompletedPayload) String 

func (x *HistoryEventPayload_ChildWorkflowCompletedPayload) String() string

type HistoryEventPayload_ChildWorkflowFailed 

type HistoryEventPayload_ChildWorkflowFailed struct {
	ChildWorkflowFailed *HistoryEventPayload_ChildWorkflowFailedPayload `protobuf:"bytes,14,opt,name=child_workflow_failed,json=childWorkflowFailed,oneof"`
}

type HistoryEventPayload_ChildWorkflowFailedPayload 

type HistoryEventPayload_ChildWorkflowFailedPayload struct {
	ChildWorkflowId []byte `protobuf:"bytes,1,opt,name=child_workflow_id,json=childWorkflowId" json:"child_workflow_id,omitempty"`
	Error           string `protobuf:"bytes,2,opt,name=error" json:"error,omitempty"`
	SystemFailure   bool   `protobuf:"varint,3,opt,name=system_failure,json=systemFailure" json:"system_failure,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_ChildWorkflowFailedPayload) Descriptor deprecated 

func (*HistoryEventPayload_ChildWorkflowFailedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_ChildWorkflowFailedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_ChildWorkflowFailedPayload) GetChildWorkflowId 

func (x *HistoryEventPayload_ChildWorkflowFailedPayload) GetChildWorkflowId() []byte

func (*HistoryEventPayload_ChildWorkflowFailedPayload) GetError 

func (x *HistoryEventPayload_ChildWorkflowFailedPayload) GetError() string

func (*HistoryEventPayload_ChildWorkflowFailedPayload) GetSystemFailure added in v19.2.0 

func (x *HistoryEventPayload_ChildWorkflowFailedPayload) GetSystemFailure() bool

func (*HistoryEventPayload_ChildWorkflowFailedPayload) ProtoMessage 

func (*HistoryEventPayload_ChildWorkflowFailedPayload) ProtoMessage()

func (*HistoryEventPayload_ChildWorkflowFailedPayload) ProtoReflect 

func (x *HistoryEventPayload_ChildWorkflowFailedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_ChildWorkflowFailedPayload) Reset 

func (x *HistoryEventPayload_ChildWorkflowFailedPayload) Reset()

func (*HistoryEventPayload_ChildWorkflowFailedPayload) String 

func (x *HistoryEventPayload_ChildWorkflowFailedPayload) String() string

type HistoryEventPayload_ChildWorkflowScheduled 

type HistoryEventPayload_ChildWorkflowScheduled struct {
	ChildWorkflowScheduled *HistoryEventPayload_ChildWorkflowScheduledPayload `protobuf:"bytes,12,opt,name=child_workflow_scheduled,json=childWorkflowScheduled,oneof"`
}

type HistoryEventPayload_ChildWorkflowScheduledPayload 

type HistoryEventPayload_ChildWorkflowScheduledPayload struct {
	ChildWorkflowName  string `protobuf:"bytes,1,opt,name=child_workflow_name,json=childWorkflowName" json:"child_workflow_name,omitempty"`
	ChildWorkflowId    []byte `protobuf:"bytes,2,opt,name=child_workflow_id,json=childWorkflowId" json:"child_workflow_id,omitempty"`
	ReplayKey          int32  `protobuf:"varint,3,opt,name=replay_key,json=replayKey" json:"replay_key,omitempty"`
	DetachedFromParent bool   `protobuf:"varint,4,opt,name=detached_from_parent,json=detachedFromParent" json:"detached_from_parent,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) Descriptor deprecated 

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_ChildWorkflowScheduledPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) GetChildWorkflowId 

func (x *HistoryEventPayload_ChildWorkflowScheduledPayload) GetChildWorkflowId() []byte

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) GetChildWorkflowName 

func (x *HistoryEventPayload_ChildWorkflowScheduledPayload) GetChildWorkflowName() string

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) GetDetachedFromParent 

func (x *HistoryEventPayload_ChildWorkflowScheduledPayload) GetDetachedFromParent() bool

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) GetReplayKey 

func (x *HistoryEventPayload_ChildWorkflowScheduledPayload) GetReplayKey() int32

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) ProtoMessage 

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) ProtoMessage()

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) ProtoReflect 

func (x *HistoryEventPayload_ChildWorkflowScheduledPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) Reset 

func (x *HistoryEventPayload_ChildWorkflowScheduledPayload) Reset()

func (*HistoryEventPayload_ChildWorkflowScheduledPayload) String 

func (x *HistoryEventPayload_ChildWorkflowScheduledPayload) String() string

type HistoryEventPayload_InlineActivityCompleted 

type HistoryEventPayload_InlineActivityCompleted struct {
	InlineActivityCompleted *HistoryEventPayload_InlineActivityCompletedPayload `protobuf:"bytes,16,opt,name=inline_activity_completed,json=inlineActivityCompleted,oneof"`
}

type HistoryEventPayload_InlineActivityCompletedPayload 

type HistoryEventPayload_InlineActivityCompletedPayload struct {
	ReplayKey    int32  `protobuf:"varint,1,opt,name=replay_key,json=replayKey" json:"replay_key,omitempty"`
	ActivityName string `protobuf:"bytes,2,opt,name=activity_name,json=activityName" json:"activity_name,omitempty"`
	Result       []byte `protobuf:"bytes,3,opt,name=result" json:"result,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_InlineActivityCompletedPayload) Descriptor deprecated 

func (*HistoryEventPayload_InlineActivityCompletedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_InlineActivityCompletedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_InlineActivityCompletedPayload) GetActivityName 

func (x *HistoryEventPayload_InlineActivityCompletedPayload) GetActivityName() string

func (*HistoryEventPayload_InlineActivityCompletedPayload) GetReplayKey 

func (x *HistoryEventPayload_InlineActivityCompletedPayload) GetReplayKey() int32

func (*HistoryEventPayload_InlineActivityCompletedPayload) GetResult 

func (x *HistoryEventPayload_InlineActivityCompletedPayload) GetResult() []byte

func (*HistoryEventPayload_InlineActivityCompletedPayload) ProtoMessage 

func (*HistoryEventPayload_InlineActivityCompletedPayload) ProtoMessage()

func (*HistoryEventPayload_InlineActivityCompletedPayload) ProtoReflect 

func (x *HistoryEventPayload_InlineActivityCompletedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_InlineActivityCompletedPayload) Reset 

func (x *HistoryEventPayload_InlineActivityCompletedPayload) Reset()

func (*HistoryEventPayload_InlineActivityCompletedPayload) String 

func (x *HistoryEventPayload_InlineActivityCompletedPayload) String() string

type HistoryEventPayload_InlineActivityFailed 

type HistoryEventPayload_InlineActivityFailed struct {
	InlineActivityFailed *HistoryEventPayload_InlineActivityFailedPayload `protobuf:"bytes,17,opt,name=inline_activity_failed,json=inlineActivityFailed,oneof"`
}

type HistoryEventPayload_InlineActivityFailedPayload 

type HistoryEventPayload_InlineActivityFailedPayload struct {
	ReplayKey     int32  `protobuf:"varint,1,opt,name=replay_key,json=replayKey" json:"replay_key,omitempty"`
	ActivityName  string `protobuf:"bytes,2,opt,name=activity_name,json=activityName" json:"activity_name,omitempty"`
	Error         string `protobuf:"bytes,3,opt,name=error" json:"error,omitempty"`
	SystemFailure bool   `protobuf:"varint,4,opt,name=system_failure,json=systemFailure" json:"system_failure,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_InlineActivityFailedPayload) Descriptor deprecated 

func (*HistoryEventPayload_InlineActivityFailedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_InlineActivityFailedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_InlineActivityFailedPayload) GetActivityName 

func (x *HistoryEventPayload_InlineActivityFailedPayload) GetActivityName() string

func (*HistoryEventPayload_InlineActivityFailedPayload) GetError 

func (x *HistoryEventPayload_InlineActivityFailedPayload) GetError() string

func (*HistoryEventPayload_InlineActivityFailedPayload) GetReplayKey 

func (x *HistoryEventPayload_InlineActivityFailedPayload) GetReplayKey() int32

func (*HistoryEventPayload_InlineActivityFailedPayload) GetSystemFailure added in v19.2.0 

func (x *HistoryEventPayload_InlineActivityFailedPayload) GetSystemFailure() bool

func (*HistoryEventPayload_InlineActivityFailedPayload) ProtoMessage 

func (*HistoryEventPayload_InlineActivityFailedPayload) ProtoMessage()

func (*HistoryEventPayload_InlineActivityFailedPayload) ProtoReflect 

func (x *HistoryEventPayload_InlineActivityFailedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_InlineActivityFailedPayload) Reset 

func (x *HistoryEventPayload_InlineActivityFailedPayload) Reset()

func (*HistoryEventPayload_InlineActivityFailedPayload) String 

func (x *HistoryEventPayload_InlineActivityFailedPayload) String() string

type HistoryEventPayload_SignalReceived 

type HistoryEventPayload_SignalReceived struct {
	SignalReceived *HistoryEventPayload_SignalReceivedPayload `protobuf:"bytes,9,opt,name=signal_received,json=signalReceived,oneof"`
}

type HistoryEventPayload_SignalReceivedPayload 

type HistoryEventPayload_SignalReceivedPayload struct {
	SignalName string `protobuf:"bytes,1,opt,name=signal_name,json=signalName" json:"signal_name,omitempty"`
	Payload    []byte `protobuf:"bytes,2,opt,name=payload" json:"payload,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_SignalReceivedPayload) Descriptor deprecated 

func (*HistoryEventPayload_SignalReceivedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_SignalReceivedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_SignalReceivedPayload) GetPayload 

func (x *HistoryEventPayload_SignalReceivedPayload) GetPayload() []byte

func (*HistoryEventPayload_SignalReceivedPayload) GetSignalName 

func (x *HistoryEventPayload_SignalReceivedPayload) GetSignalName() string

func (*HistoryEventPayload_SignalReceivedPayload) ProtoMessage 

func (*HistoryEventPayload_SignalReceivedPayload) ProtoMessage()

func (*HistoryEventPayload_SignalReceivedPayload) ProtoReflect 

func (x *HistoryEventPayload_SignalReceivedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_SignalReceivedPayload) Reset 

func (x *HistoryEventPayload_SignalReceivedPayload) Reset()

func (*HistoryEventPayload_SignalReceivedPayload) String 

func (x *HistoryEventPayload_SignalReceivedPayload) String() string

type HistoryEventPayload_TimerFired 

type HistoryEventPayload_TimerFired struct {
	TimerFired *HistoryEventPayload_TimerFiredPayload `protobuf:"bytes,8,opt,name=timer_fired,json=timerFired,oneof"`
}

type HistoryEventPayload_TimerFiredPayload 

type HistoryEventPayload_TimerFiredPayload struct {
	ScheduledEventSequenceId int64 `` /* 131-byte string literal not displayed */
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_TimerFiredPayload) Descriptor deprecated 

func (*HistoryEventPayload_TimerFiredPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_TimerFiredPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_TimerFiredPayload) GetScheduledEventSequenceId 

func (x *HistoryEventPayload_TimerFiredPayload) GetScheduledEventSequenceId() int64

func (*HistoryEventPayload_TimerFiredPayload) ProtoMessage 

func (*HistoryEventPayload_TimerFiredPayload) ProtoMessage()

func (*HistoryEventPayload_TimerFiredPayload) ProtoReflect 

func (x *HistoryEventPayload_TimerFiredPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_TimerFiredPayload) Reset 

func (x *HistoryEventPayload_TimerFiredPayload) Reset()

func (*HistoryEventPayload_TimerFiredPayload) String 

func (x *HistoryEventPayload_TimerFiredPayload) String() string

type HistoryEventPayload_TimerStarted 

type HistoryEventPayload_TimerStarted struct {
	TimerStarted *HistoryEventPayload_TimerStartedPayload `protobuf:"bytes,7,opt,name=timer_started,json=timerStarted,oneof"`
}

type HistoryEventPayload_TimerStartedPayload 

type HistoryEventPayload_TimerStartedPayload struct {
	Duration  *durationpb.Duration `protobuf:"bytes,1,opt,name=duration" json:"duration,omitempty"`
	ReplayKey int32                `protobuf:"varint,2,opt,name=replay_key,json=replayKey" json:"replay_key,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_TimerStartedPayload) Descriptor deprecated 

func (*HistoryEventPayload_TimerStartedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_TimerStartedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_TimerStartedPayload) GetDuration 

func (x *HistoryEventPayload_TimerStartedPayload) GetDuration() *durationpb.Duration

func (*HistoryEventPayload_TimerStartedPayload) GetReplayKey 

func (x *HistoryEventPayload_TimerStartedPayload) GetReplayKey() int32

func (*HistoryEventPayload_TimerStartedPayload) ProtoMessage 

func (*HistoryEventPayload_TimerStartedPayload) ProtoMessage()

func (*HistoryEventPayload_TimerStartedPayload) ProtoReflect 

func (x *HistoryEventPayload_TimerStartedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_TimerStartedPayload) Reset 

func (x *HistoryEventPayload_TimerStartedPayload) Reset()

func (*HistoryEventPayload_TimerStartedPayload) String 

func (x *HistoryEventPayload_TimerStartedPayload) String() string

type HistoryEventPayload_WorkflowCanceled 

type HistoryEventPayload_WorkflowCanceled struct {
	WorkflowCanceled *HistoryEventPayload_WorkflowCanceledPayload `protobuf:"bytes,11,opt,name=workflow_canceled,json=workflowCanceled,oneof"`
}

type HistoryEventPayload_WorkflowCanceledPayload 

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

func (*HistoryEventPayload_WorkflowCanceledPayload) Descriptor deprecated 

func (*HistoryEventPayload_WorkflowCanceledPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_WorkflowCanceledPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_WorkflowCanceledPayload) ProtoMessage 

func (*HistoryEventPayload_WorkflowCanceledPayload) ProtoMessage()

func (*HistoryEventPayload_WorkflowCanceledPayload) ProtoReflect 

func (x *HistoryEventPayload_WorkflowCanceledPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_WorkflowCanceledPayload) Reset 

func (x *HistoryEventPayload_WorkflowCanceledPayload) Reset()

func (*HistoryEventPayload_WorkflowCanceledPayload) String 

func (x *HistoryEventPayload_WorkflowCanceledPayload) String() string

type HistoryEventPayload_WorkflowCancellationRequested 

type HistoryEventPayload_WorkflowCancellationRequested struct {
	WorkflowCancellationRequested *HistoryEventPayload_WorkflowCancellationRequestedPayload `protobuf:"bytes,10,opt,name=workflow_cancellation_requested,json=workflowCancellationRequested,oneof"`
}

type HistoryEventPayload_WorkflowCancellationRequestedPayload 

type HistoryEventPayload_WorkflowCancellationRequestedPayload struct {
	Reason CancellationReason `protobuf:"varint,1,opt,name=reason,enum=gitlab.agent.autocore.CancellationReason" json:"reason,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) Descriptor deprecated 

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_WorkflowCancellationRequestedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) GetReason added in v19.2.0 

func (x *HistoryEventPayload_WorkflowCancellationRequestedPayload) GetReason() CancellationReason

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) ProtoMessage 

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) ProtoMessage()

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) ProtoReflect 

func (x *HistoryEventPayload_WorkflowCancellationRequestedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) Reset 

func (x *HistoryEventPayload_WorkflowCancellationRequestedPayload) Reset()

func (*HistoryEventPayload_WorkflowCancellationRequestedPayload) String 

func (x *HistoryEventPayload_WorkflowCancellationRequestedPayload) String() string

type HistoryEventPayload_WorkflowCompleted 

type HistoryEventPayload_WorkflowCompleted struct {
	WorkflowCompleted *HistoryEventPayload_WorkflowCompletedPayload `protobuf:"bytes,2,opt,name=workflow_completed,json=workflowCompleted,oneof"`
}

type HistoryEventPayload_WorkflowCompletedPayload 

type HistoryEventPayload_WorkflowCompletedPayload struct {
	Result []byte `protobuf:"bytes,1,opt,name=result" json:"result,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_WorkflowCompletedPayload) Descriptor deprecated 

func (*HistoryEventPayload_WorkflowCompletedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_WorkflowCompletedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_WorkflowCompletedPayload) GetResult 

func (x *HistoryEventPayload_WorkflowCompletedPayload) GetResult() []byte

func (*HistoryEventPayload_WorkflowCompletedPayload) ProtoMessage 

func (*HistoryEventPayload_WorkflowCompletedPayload) ProtoMessage()

func (*HistoryEventPayload_WorkflowCompletedPayload) ProtoReflect 

func (x *HistoryEventPayload_WorkflowCompletedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_WorkflowCompletedPayload) Reset 

func (x *HistoryEventPayload_WorkflowCompletedPayload) Reset()

func (*HistoryEventPayload_WorkflowCompletedPayload) String 

func (x *HistoryEventPayload_WorkflowCompletedPayload) String() string

type HistoryEventPayload_WorkflowCreated 

type HistoryEventPayload_WorkflowCreated struct {
	WorkflowCreated *HistoryEventPayload_WorkflowCreatedPayload `protobuf:"bytes,1,opt,name=workflow_created,json=workflowCreated,oneof"`
}

type HistoryEventPayload_WorkflowCreatedPayload 

type HistoryEventPayload_WorkflowCreatedPayload struct {
	Name  string `protobuf:"bytes,1,opt,name=name" json:"name,omitempty"`
	Input []byte `protobuf:"bytes,2,opt,name=input" json:"input,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_WorkflowCreatedPayload) Descriptor deprecated 

func (*HistoryEventPayload_WorkflowCreatedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_WorkflowCreatedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_WorkflowCreatedPayload) GetInput 

func (x *HistoryEventPayload_WorkflowCreatedPayload) GetInput() []byte

func (*HistoryEventPayload_WorkflowCreatedPayload) GetName 

func (x *HistoryEventPayload_WorkflowCreatedPayload) GetName() string

func (*HistoryEventPayload_WorkflowCreatedPayload) ProtoMessage 

func (*HistoryEventPayload_WorkflowCreatedPayload) ProtoMessage()

func (*HistoryEventPayload_WorkflowCreatedPayload) ProtoReflect 

func (x *HistoryEventPayload_WorkflowCreatedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_WorkflowCreatedPayload) Reset 

func (x *HistoryEventPayload_WorkflowCreatedPayload) Reset()

func (*HistoryEventPayload_WorkflowCreatedPayload) String 

func (x *HistoryEventPayload_WorkflowCreatedPayload) String() string

type HistoryEventPayload_WorkflowFailed 

type HistoryEventPayload_WorkflowFailed struct {
	WorkflowFailed *HistoryEventPayload_WorkflowFailedPayload `protobuf:"bytes,3,opt,name=workflow_failed,json=workflowFailed,oneof"`
}

type HistoryEventPayload_WorkflowFailedPayload 

type HistoryEventPayload_WorkflowFailedPayload struct {
	Error string `protobuf:"bytes,1,opt,name=error" json:"error,omitempty"`
	// contains filtered or unexported fields
}

func (*HistoryEventPayload_WorkflowFailedPayload) Descriptor deprecated 

func (*HistoryEventPayload_WorkflowFailedPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_WorkflowFailedPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_WorkflowFailedPayload) GetError 

func (x *HistoryEventPayload_WorkflowFailedPayload) GetError() string

func (*HistoryEventPayload_WorkflowFailedPayload) ProtoMessage 

func (*HistoryEventPayload_WorkflowFailedPayload) ProtoMessage()

func (*HistoryEventPayload_WorkflowFailedPayload) ProtoReflect 

func (x *HistoryEventPayload_WorkflowFailedPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_WorkflowFailedPayload) Reset 

func (x *HistoryEventPayload_WorkflowFailedPayload) Reset()

func (*HistoryEventPayload_WorkflowFailedPayload) String 

func (x *HistoryEventPayload_WorkflowFailedPayload) String() string

type HistoryEventPayload_WorkflowTimedOut added in v19.2.0 

type HistoryEventPayload_WorkflowTimedOut struct {
	WorkflowTimedOut *HistoryEventPayload_WorkflowTimedOutPayload `protobuf:"bytes,18,opt,name=workflow_timed_out,json=workflowTimedOut,oneof"`
}

type HistoryEventPayload_WorkflowTimedOutPayload added in v19.2.0 

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

func (*HistoryEventPayload_WorkflowTimedOutPayload) Descriptor deprecated added in v19.2.0 

func (*HistoryEventPayload_WorkflowTimedOutPayload) Descriptor() ([]byte, []int)

Deprecated: Use HistoryEventPayload_WorkflowTimedOutPayload.ProtoReflect.Descriptor instead.

func (*HistoryEventPayload_WorkflowTimedOutPayload) ProtoMessage added in v19.2.0 

func (*HistoryEventPayload_WorkflowTimedOutPayload) ProtoMessage()

func (*HistoryEventPayload_WorkflowTimedOutPayload) ProtoReflect added in v19.2.0 

func (x *HistoryEventPayload_WorkflowTimedOutPayload) ProtoReflect() protoreflect.Message

func (*HistoryEventPayload_WorkflowTimedOutPayload) Reset added in v19.2.0 

func (x *HistoryEventPayload_WorkflowTimedOutPayload) Reset()

func (*HistoryEventPayload_WorkflowTimedOutPayload) String added in v19.2.0 

func (x *HistoryEventPayload_WorkflowTimedOutPayload) String() string

type HistoryEventRow 

type HistoryEventRow struct {
	ShardID    ShardID
	WorkflowID WorkflowID
	SequenceID SequenceID
	Payload    *HistoryEventPayload
	CreatedAt  time.Time
}

HistoryEventRow is the read model scanned from history_event. Its Payload is decoded into a HistoryEventPayload during the scan.

func ListHistoryEvents 

func ListHistoryEvents(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID) ([]HistoryEventRow, error)

func ListHistoryEventsAfter 

func ListHistoryEventsAfter(ctx context.Context, tx pgx.Tx, shardID ShardID, workflowID WorkflowID, afterSeqID SequenceID) ([]HistoryEventRow, error)

type NamespaceID 

type NamespaceID int64

NamespaceID is autocore's internal namespace surrogate (namespace.id). It is the global cross-database key stored in workflow_identity.namespace_id.

type OwnerID 

type OwnerID int64

type PartitionStat added in v19.2.0 

type PartitionStat struct {
	Table             string
	Count             int64
	NewestBucketStart time.Time // zero when the table has no partitions
}

PartitionStat summarizes the partitions of one retained table.

func PartitionStats added in v19.2.0 

func PartitionStats(ctx context.Context, db DB) ([]PartitionStat, error)

PartitionStats returns the partition count and newest bucket start for each retained table.

type PollSpec added in v19.2.0 

type PollSpec struct {
	PredicateConfig []byte                 `protobuf:"bytes,1,opt,name=predicate_config,json=predicateConfig" json:"predicate_config,omitempty"`
	Interval        *durationpb.Duration   `protobuf:"bytes,2,opt,name=interval" json:"interval,omitempty"`
	Deadline        *timestamppb.Timestamp `protobuf:"bytes,3,opt,name=deadline" json:"deadline,omitempty"`
	// contains filtered or unexported fields
}

func (*PollSpec) Descriptor deprecated added in v19.2.0 

func (*PollSpec) Descriptor() ([]byte, []int)

Deprecated: Use PollSpec.ProtoReflect.Descriptor instead.

func (*PollSpec) GetDeadline added in v19.2.0 

func (x *PollSpec) GetDeadline() *timestamppb.Timestamp

func (*PollSpec) GetInterval added in v19.2.0 

func (x *PollSpec) GetInterval() *durationpb.Duration

func (*PollSpec) GetPredicateConfig added in v19.2.0 

func (x *PollSpec) GetPredicateConfig() []byte

func (*PollSpec) ProtoMessage added in v19.2.0 

func (*PollSpec) ProtoMessage()

func (*PollSpec) ProtoReflect added in v19.2.0 

func (x *PollSpec) ProtoReflect() protoreflect.Message

func (*PollSpec) Reset added in v19.2.0 

func (x *PollSpec) Reset()

func (*PollSpec) String added in v19.2.0 

func (x *PollSpec) String() string

type RetryPolicy 

type RetryPolicy struct {
	MaxAttempts        int32                `protobuf:"varint,1,opt,name=max_attempts,json=maxAttempts" json:"max_attempts,omitempty"`
	InitialInterval    *durationpb.Duration `protobuf:"bytes,2,opt,name=initial_interval,json=initialInterval" json:"initial_interval,omitempty"`
	BackoffCoefficient float64              `protobuf:"fixed64,3,opt,name=backoff_coefficient,json=backoffCoefficient" json:"backoff_coefficient,omitempty"`
	MaxInterval        *durationpb.Duration `protobuf:"bytes,4,opt,name=max_interval,json=maxInterval" json:"max_interval,omitempty"`
	// contains filtered or unexported fields
}

func (*RetryPolicy) Descriptor deprecated 

func (*RetryPolicy) Descriptor() ([]byte, []int)

Deprecated: Use RetryPolicy.ProtoReflect.Descriptor instead.

func (*RetryPolicy) GetBackoffCoefficient 

func (x *RetryPolicy) GetBackoffCoefficient() float64

func (*RetryPolicy) GetInitialInterval 

func (x *RetryPolicy) GetInitialInterval() *durationpb.Duration

func (*RetryPolicy) GetMaxAttempts 

func (x *RetryPolicy) GetMaxAttempts() int32

func (*RetryPolicy) GetMaxInterval 

func (x *RetryPolicy) GetMaxInterval() *durationpb.Duration

func (*RetryPolicy) ProtoMessage 

func (*RetryPolicy) ProtoMessage()

func (*RetryPolicy) ProtoReflect 

func (x *RetryPolicy) ProtoReflect() protoreflect.Message

func (*RetryPolicy) Reset 

func (x *RetryPolicy) Reset()

func (*RetryPolicy) String 

func (x *RetryPolicy) String() string

type ScheduledTaskID 

type ScheduledTaskID uuid.UUID //nolint: recvcheck

ScheduledTaskID identifies a scheduled task (timer or retry).

func NewScheduledTaskID added in v19.2.0 

func NewScheduledTaskID() ScheduledTaskID

func (ScheduledTaskID) Bytes 

func (id ScheduledTaskID) Bytes() []byte

Bytes returns the UUID as a byte slice, for use in protobuf fields.

func (*ScheduledTaskID) ScanUUID 

func (id *ScheduledTaskID) ScanUUID(v pgtype.UUID) error

func (ScheduledTaskID) String 

func (id ScheduledTaskID) String() string

func (ScheduledTaskID) UUIDValue 

func (id ScheduledTaskID) UUIDValue() (pgtype.UUID, error)

type ScheduledTaskInsert 

type ScheduledTaskInsert struct {
	ShardID         ShardID
	WorkflowID      WorkflowID
	ScheduledTaskID ScheduledTaskID
	Payload         []byte
}

ScheduledTaskInsert holds the columns written when creating a scheduled_task. The fire time is supplied separately as a fire-after duration, not a column.

type ScheduledTaskPayload 

type ScheduledTaskPayload struct {

	// Types that are valid to be assigned to Task:
	//
	//	*ScheduledTaskPayload_Timer
	//	*ScheduledTaskPayload_ActivityRetry
	//	*ScheduledTaskPayload_WorkflowRetry
	//	*ScheduledTaskPayload_WorkflowTimeout
	Task isScheduledTaskPayload_Task `protobuf_oneof:"task"`
	// contains filtered or unexported fields
}

func (*ScheduledTaskPayload) Descriptor deprecated 

func (*ScheduledTaskPayload) Descriptor() ([]byte, []int)

Deprecated: Use ScheduledTaskPayload.ProtoReflect.Descriptor instead.

func (*ScheduledTaskPayload) GetActivityRetry 

func (x *ScheduledTaskPayload) GetActivityRetry() *ScheduledTaskPayload_ActivityRetryPayload

func (*ScheduledTaskPayload) GetTask 

func (x *ScheduledTaskPayload) GetTask() isScheduledTaskPayload_Task

func (*ScheduledTaskPayload) GetTimer 

func (x *ScheduledTaskPayload) GetTimer() *ScheduledTaskPayload_TimerPayload

func (*ScheduledTaskPayload) GetWorkflowRetry 

func (x *ScheduledTaskPayload) GetWorkflowRetry() *ScheduledTaskPayload_WorkflowRetryPayload

func (*ScheduledTaskPayload) GetWorkflowTimeout added in v19.2.0 

func (x *ScheduledTaskPayload) GetWorkflowTimeout() *ScheduledTaskPayload_WorkflowTimeoutPayload

func (*ScheduledTaskPayload) ProtoMessage 

func (*ScheduledTaskPayload) ProtoMessage()

func (*ScheduledTaskPayload) ProtoReflect 

func (x *ScheduledTaskPayload) ProtoReflect() protoreflect.Message

func (*ScheduledTaskPayload) Reset 

func (x *ScheduledTaskPayload) Reset()

func (*ScheduledTaskPayload) ScanBytes 

func (x *ScheduledTaskPayload) ScanBytes(src []byte) error

func (*ScheduledTaskPayload) String 

func (x *ScheduledTaskPayload) String() string

type ScheduledTaskPayload_ActivityRetry 

type ScheduledTaskPayload_ActivityRetry struct {
	ActivityRetry *ScheduledTaskPayload_ActivityRetryPayload `protobuf:"bytes,2,opt,name=activity_retry,json=activityRetry,oneof"`
}

type ScheduledTaskPayload_ActivityRetryPayload 

type ScheduledTaskPayload_ActivityRetryPayload struct {
	TaskId []byte `protobuf:"bytes,1,opt,name=task_id,json=taskId" json:"task_id,omitempty"`
	// contains filtered or unexported fields
}

func (*ScheduledTaskPayload_ActivityRetryPayload) Descriptor deprecated 

func (*ScheduledTaskPayload_ActivityRetryPayload) Descriptor() ([]byte, []int)

Deprecated: Use ScheduledTaskPayload_ActivityRetryPayload.ProtoReflect.Descriptor instead.

func (*ScheduledTaskPayload_ActivityRetryPayload) GetTaskId 

func (x *ScheduledTaskPayload_ActivityRetryPayload) GetTaskId() []byte

func (*ScheduledTaskPayload_ActivityRetryPayload) ProtoMessage 

func (*ScheduledTaskPayload_ActivityRetryPayload) ProtoMessage()

func (*ScheduledTaskPayload_ActivityRetryPayload) ProtoReflect 

func (x *ScheduledTaskPayload_ActivityRetryPayload) ProtoReflect() protoreflect.Message

func (*ScheduledTaskPayload_ActivityRetryPayload) Reset 

func (x *ScheduledTaskPayload_ActivityRetryPayload) Reset()

func (*ScheduledTaskPayload_ActivityRetryPayload) String 

func (x *ScheduledTaskPayload_ActivityRetryPayload) String() string

type ScheduledTaskPayload_Timer 

type ScheduledTaskPayload_Timer struct {
	Timer *ScheduledTaskPayload_TimerPayload `protobuf:"bytes,1,opt,name=timer,oneof"`
}

type ScheduledTaskPayload_TimerPayload 

type ScheduledTaskPayload_TimerPayload struct {
	ScheduledEventSequenceId int64 `` /* 131-byte string literal not displayed */
	// contains filtered or unexported fields
}

func (*ScheduledTaskPayload_TimerPayload) Descriptor deprecated 

func (*ScheduledTaskPayload_TimerPayload) Descriptor() ([]byte, []int)

Deprecated: Use ScheduledTaskPayload_TimerPayload.ProtoReflect.Descriptor instead.

func (*ScheduledTaskPayload_TimerPayload) GetScheduledEventSequenceId 

func (x *ScheduledTaskPayload_TimerPayload) GetScheduledEventSequenceId() int64

func (*ScheduledTaskPayload_TimerPayload) ProtoMessage 

func (*ScheduledTaskPayload_TimerPayload) ProtoMessage()

func (*ScheduledTaskPayload_TimerPayload) ProtoReflect 

func (x *ScheduledTaskPayload_TimerPayload) ProtoReflect() protoreflect.Message

func (*ScheduledTaskPayload_TimerPayload) Reset 

func (x *ScheduledTaskPayload_TimerPayload) Reset()

func (*ScheduledTaskPayload_TimerPayload) String 

func (x *ScheduledTaskPayload_TimerPayload) String() string

type ScheduledTaskPayload_WorkflowRetry 

type ScheduledTaskPayload_WorkflowRetry struct {
	WorkflowRetry *ScheduledTaskPayload_WorkflowRetryPayload `protobuf:"bytes,3,opt,name=workflow_retry,json=workflowRetry,oneof"`
}

type ScheduledTaskPayload_WorkflowRetryPayload 

type ScheduledTaskPayload_WorkflowRetryPayload struct {
	TaskId []byte `protobuf:"bytes,1,opt,name=task_id,json=taskId" json:"task_id,omitempty"`
	// contains filtered or unexported fields
}

func (*ScheduledTaskPayload_WorkflowRetryPayload) Descriptor deprecated 

func (*ScheduledTaskPayload_WorkflowRetryPayload) Descriptor() ([]byte, []int)

Deprecated: Use ScheduledTaskPayload_WorkflowRetryPayload.ProtoReflect.Descriptor instead.

func (*ScheduledTaskPayload_WorkflowRetryPayload) GetTaskId 

func (x *ScheduledTaskPayload_WorkflowRetryPayload) GetTaskId() []byte

func (*ScheduledTaskPayload_WorkflowRetryPayload) ProtoMessage 

func (*ScheduledTaskPayload_WorkflowRetryPayload) ProtoMessage()

func (*ScheduledTaskPayload_WorkflowRetryPayload) ProtoReflect 

func (x *ScheduledTaskPayload_WorkflowRetryPayload) ProtoReflect() protoreflect.Message

func (*ScheduledTaskPayload_WorkflowRetryPayload) Reset 

func (x *ScheduledTaskPayload_WorkflowRetryPayload) Reset()

func (*ScheduledTaskPayload_WorkflowRetryPayload) String 

func (x *ScheduledTaskPayload_WorkflowRetryPayload) String() string

type ScheduledTaskPayload_WorkflowTimeout added in v19.2.0 

type ScheduledTaskPayload_WorkflowTimeout struct {
	WorkflowTimeout *ScheduledTaskPayload_WorkflowTimeoutPayload `protobuf:"bytes,4,opt,name=workflow_timeout,json=workflowTimeout,oneof"`
}

type ScheduledTaskPayload_WorkflowTimeoutPayload added in v19.2.0 

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

func (*ScheduledTaskPayload_WorkflowTimeoutPayload) Descriptor deprecated added in v19.2.0 

func (*ScheduledTaskPayload_WorkflowTimeoutPayload) Descriptor() ([]byte, []int)

Deprecated: Use ScheduledTaskPayload_WorkflowTimeoutPayload.ProtoReflect.Descriptor instead.

func (*ScheduledTaskPayload_WorkflowTimeoutPayload) ProtoMessage added in v19.2.0 

func (*ScheduledTaskPayload_WorkflowTimeoutPayload) ProtoMessage()

func (*ScheduledTaskPayload_WorkflowTimeoutPayload) ProtoReflect added in v19.2.0 

func (x *ScheduledTaskPayload_WorkflowTimeoutPayload) ProtoReflect() protoreflect.Message

func (*ScheduledTaskPayload_WorkflowTimeoutPayload) Reset added in v19.2.0 

func (x *ScheduledTaskPayload_WorkflowTimeoutPayload) Reset()

func (*ScheduledTaskPayload_WorkflowTimeoutPayload) String added in v19.2.0 

func (x *ScheduledTaskPayload_WorkflowTimeoutPayload) String() string

type ScheduledTaskResult 

type ScheduledTaskResult struct {
	ScheduledTaskID ScheduledTaskID
	FireAt          time.Time
	Remaining       time.Duration
}

ScheduledTaskResult is the per-row outcome of CreateScheduledTasks: the DB-computed fire_at and the remaining duration until it, keyed by ScheduledTaskID so callers can match results back to input rows regardless of RETURNING order.

func CreateScheduledTasks 

func CreateScheduledTasks(ctx context.Context, tx pgx.Tx, rows []*ScheduledTaskInsert, fireAfters []time.Duration) ([]ScheduledTaskResult, error)

CreateScheduledTasks inserts scheduled tasks that each fire after the corresponding fireAfters duration, computed relative to the DB clock (now() + fireAfter). fireAfters is parallel to rows. For each row it returns the DB-computed fire_at and the remaining duration until fire_at as measured by the DB's wall clock (clock_timestamp), which accounts for transaction latency and clock skew between app and DB. Result order is not guaranteed to match input order; match on ScheduledTaskID.

type ScheduledTaskRow 

type ScheduledTaskRow struct {
	ShardID         ShardID
	WorkflowID      WorkflowID
	ScheduledTaskID ScheduledTaskID
	Payload         *ScheduledTaskPayload
	FireAt          time.Time
	CreatedAt       time.Time
}

ScheduledTaskRow is the read model scanned from scheduled_task. Its Payload is decoded into a ScheduledTaskPayload during the scan.

func FindDueScheduledTasks 

func FindDueScheduledTasks(ctx context.Context, tx pgx.Tx, shardID ShardID, limit int) ([]ScheduledTaskRow, error)

type SequenceID 

type SequenceID int64

type Shard 

type Shard struct {
	ShardID        ShardID
	OwnerID        OwnerID
	FenceID        FenceID
	LeaseExpiresAt time.Time
	State          ShardState
}

type ShardID 

type ShardID int32

func ListShardsWithPendingActivityTasks 

func ListShardsWithPendingActivityTasks(ctx context.Context, tx pgx.Tx, shardIDs []ShardID) ([]ShardID, error)

type ShardListing 

type ShardListing struct {
	Shards []Shard
	// TODO: not sure if this actually turns out to be useful, I left it in to remember this.
	// Basically, we can have clock skew between KAS pods and the database clock and when
	// assessing the leases this matters. Let's see ..
	DBNow time.Time
}

type ShardMappingRow 

type ShardMappingRow struct {
	ShardID ShardID
	DBID    DBID
}

ShardMappingRow represents a shard-to-database mapping from the central database's shard table.

type ShardState 

type ShardState int16

ShardState represents the lifecycle state of a shard lease. Important: do not change constants! They are used in DB data. 

const (
	ShardStateUnknown  ShardState = 0
	ShardStateActive   ShardState = 1
	ShardStateDraining ShardState = 2
)

func (ShardState) String 

func (s ShardState) String() string

type SignalID 

type SignalID uuid.UUID //nolint: recvcheck

SignalID identifies a signal sent to a workflow.

func NewSignalID added in v19.2.0 

func NewSignalID() SignalID

func (SignalID) Bytes 

func (id SignalID) Bytes() []byte

Bytes returns the UUID as a byte slice, for use in protobuf fields.

func (*SignalID) ScanUUID 

func (id *SignalID) ScanUUID(v pgtype.UUID) error

func (SignalID) String 

func (id SignalID) String() string

func (SignalID) UUIDValue 

func (id SignalID) UUIDValue() (pgtype.UUID, error)

type SignalRow 

type SignalRow struct {
	ShardID     ShardID
	WorkflowID  WorkflowID
	SignalID    SignalID
	IdentityKey string
	SignalType  SignalType
	SignalName  string
	Payload     []byte
	CreatedAt   time.Time
}

func ConsumePendingSignals 

func ConsumePendingSignals(ctx context.Context, tx pgx.Tx, shardID ShardID, limit int) ([]SignalRow, error)

ConsumePendingSignals atomically deletes up to limit pending inbox rows for the shard, choosing the batch oldest-first by created_at, and returns them ordered by (workflow_id, created_at). The workflow_id ordering makes the scanner take FOR UPDATE on workflow_execution rows in workflow_id order, matching the timer scan so the two cannot deadlock; created_at remains the secondary key so one workflow's signals are still promoted in arrival order. The signal_identity rows are left in place so re-sends stay deduplicated. The limit bounds the caller's fenced transaction; callers loop until a short batch signals the inbox is drained. Intended to be called inside a WithSharedShardFence transaction by the scanner.

type SignalType 

type SignalType int16

SignalType identifies what kind of signal is stored in the signal_inbox table. The signal scanner uses this to determine which history event type to emit when promoting the signal. 

const (
	SignalTypeUnknown        SignalType = 0
	SignalTypeCancelWorkflow SignalType = 1
	SignalTypeChannel        SignalType = 2
	// SignalTypeChildTerminal carries a child workflow's terminal outcome to its
	// parent's inbox.
	SignalTypeChildTerminal SignalType = 3
)

func (SignalType) String 

func (s SignalType) String() string

type SpanContext 

type SpanContext struct {
	TraceId    []byte `protobuf:"bytes,1,opt,name=trace_id,json=traceId" json:"trace_id,omitempty"`
	SpanId     []byte `protobuf:"bytes,2,opt,name=span_id,json=spanId" json:"span_id,omitempty"`
	TraceFlags uint32 `protobuf:"varint,3,opt,name=trace_flags,json=traceFlags" json:"trace_flags,omitempty"`
	// contains filtered or unexported fields
}

func (*SpanContext) Descriptor deprecated 

func (*SpanContext) Descriptor() ([]byte, []int)

Deprecated: Use SpanContext.ProtoReflect.Descriptor instead.

func (*SpanContext) GetSpanId 

func (x *SpanContext) GetSpanId() []byte

func (*SpanContext) GetTraceFlags 

func (x *SpanContext) GetTraceFlags() uint32

func (*SpanContext) GetTraceId 

func (x *SpanContext) GetTraceId() []byte

func (*SpanContext) ProtoMessage 

func (*SpanContext) ProtoMessage()

func (*SpanContext) ProtoReflect 

func (x *SpanContext) ProtoReflect() protoreflect.Message

func (*SpanContext) Reset 

func (x *SpanContext) Reset()

func (*SpanContext) String 

func (x *SpanContext) String() string

type SystemStore 

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

SystemStore owns shard-lifecycle queries. It wraps a dedicated pgxpool.Pool that must not be shared with workload queries so heartbeat latency is independent of workload pool queueing.

func NewSystemStore 

func NewSystemStore(pool *pgxpool.Pool) *SystemStore

func (*SystemStore) ClaimShards 

func (s *SystemStore) ClaimShards(ctx context.Context, ownerID OwnerID, leaseDuration time.Duration, limit int) ([]ClaimedShard, error)

func (*SystemStore) EnsureShards 

func (s *SystemStore) EnsureShards(ctx context.Context, desiredCount int) error

func (*SystemStore) EnsureSpecificShards 

func (s *SystemStore) EnsureSpecificShards(ctx context.Context, shardIDs []ShardID) error

func (*SystemStore) HeartbeatShard 

func (s *SystemStore) HeartbeatShard(ctx context.Context, shardID ShardID, ownerID OwnerID, fenceID FenceID, leaseDuration time.Duration) (bool, error)

func (*SystemStore) ListShards 

func (s *SystemStore) ListShards(ctx context.Context) (*ShardListing, error)

func (*SystemStore) MarkShardDraining 

func (s *SystemStore) MarkShardDraining(ctx context.Context, shardID ShardID, ownerID OwnerID, fenceID FenceID) (bool, error)

func (*SystemStore) ReleaseShard 

func (s *SystemStore) ReleaseShard(ctx context.Context, shardID ShardID, ownerID OwnerID, fenceID FenceID) (bool, error)

type WorkflowDatabaseRow 

type WorkflowDatabaseRow struct {
	DBID         DBID
	MigrationDSN string
	RuntimeDSN   string
}

WorkflowDatabaseRow represents a row in the central database's workflow_database table.

MigrationDSN is used once at startup to run schema migrations and may carry elevated privileges. RuntimeDSN is used for the long-lived connection pools that serve normal workflow traffic.

type WorkflowDatabaseWithStats 

type WorkflowDatabaseWithStats struct {
	WorkflowDatabaseRow
	ShardCount int
	MinShard   ShardID
	MaxShard   ShardID
}

WorkflowDatabaseWithStats is a workflow_database row together with a summary of its shard allocation (count and inclusive shard-id range). MinShard and MaxShard are 0 when the database has no shards.

type WorkflowExecutionInsert 

type WorkflowExecutionInsert struct {
	ShardID            ShardID
	WorkflowID         WorkflowID
	Name               string
	IdentityKey        string
	State              WorkflowExecutionState
	Payload            []byte
	NextSequenceID     SequenceID
	ParentWorkflowID   *WorkflowID
	DetachedFromParent bool
}

WorkflowExecutionInsert holds the columns written when creating a workflow_execution. Payload is the already-marshaled WorkflowExecutionPayload.

type WorkflowExecutionPayload 

type WorkflowExecutionPayload struct {
	Observability *WorkflowExecutionPayload_Observability `protobuf:"bytes,1,opt,name=observability" json:"observability,omitempty"`
	Metadata      []byte                                  `protobuf:"bytes,2,opt,name=metadata" json:"metadata,omitempty"`
	// contains filtered or unexported fields
}

func (*WorkflowExecutionPayload) Descriptor deprecated 

func (*WorkflowExecutionPayload) Descriptor() ([]byte, []int)

Deprecated: Use WorkflowExecutionPayload.ProtoReflect.Descriptor instead.

func (*WorkflowExecutionPayload) GetMetadata added in v19.2.0 

func (x *WorkflowExecutionPayload) GetMetadata() []byte

func (*WorkflowExecutionPayload) GetObservability 

func (x *WorkflowExecutionPayload) GetObservability() *WorkflowExecutionPayload_Observability

func (*WorkflowExecutionPayload) ProtoMessage 

func (*WorkflowExecutionPayload) ProtoMessage()

func (*WorkflowExecutionPayload) ProtoReflect 

func (x *WorkflowExecutionPayload) ProtoReflect() protoreflect.Message

func (*WorkflowExecutionPayload) Reset 

func (x *WorkflowExecutionPayload) Reset()

func (*WorkflowExecutionPayload) ScanBytes 

func (x *WorkflowExecutionPayload) ScanBytes(src []byte) error

func (*WorkflowExecutionPayload) String 

func (x *WorkflowExecutionPayload) String() string

type WorkflowExecutionPayload_Observability 

type WorkflowExecutionPayload_Observability struct {
	Parent *SpanContext `protobuf:"bytes,1,opt,name=parent" json:"parent,omitempty"`
	// contains filtered or unexported fields
}

func (*WorkflowExecutionPayload_Observability) Descriptor deprecated 

func (*WorkflowExecutionPayload_Observability) Descriptor() ([]byte, []int)

Deprecated: Use WorkflowExecutionPayload_Observability.ProtoReflect.Descriptor instead.

func (*WorkflowExecutionPayload_Observability) GetParent 

func (x *WorkflowExecutionPayload_Observability) GetParent() *SpanContext

func (*WorkflowExecutionPayload_Observability) ProtoMessage 

func (*WorkflowExecutionPayload_Observability) ProtoMessage()

func (*WorkflowExecutionPayload_Observability) ProtoReflect 

func (x *WorkflowExecutionPayload_Observability) ProtoReflect() protoreflect.Message

func (*WorkflowExecutionPayload_Observability) Reset 

func (x *WorkflowExecutionPayload_Observability) Reset()

func (*WorkflowExecutionPayload_Observability) String 

func (x *WorkflowExecutionPayload_Observability) String() string

type WorkflowExecutionRow 

type WorkflowExecutionRow struct {
	ShardID            ShardID
	WorkflowID         WorkflowID
	Name               string
	IdentityKey        string
	State              WorkflowExecutionState
	Revision           int64
	NextSequenceID     SequenceID
	SignalEventCount   int32
	ParentWorkflowID   *WorkflowID // nil for top-level workflows
	DetachedFromParent bool        // TRUE for fire-and-forget children whose terminal events are not routed back to the parent
	CreatedAt          time.Time
	UpdatedAt          time.Time
}

WorkflowExecutionRow is the read model scanned from workflow_execution without its payload. Use WorkflowExecutionRowWithPayload when the WorkflowExecutionPayload is needed; keeping the payload off this type makes accidental payload access on a payload-less read a compile error.

func LockAndGetWorkflowExecution 

func LockAndGetWorkflowExecution(ctx context.Context, tx pgx.Tx, shardID ShardID, id WorkflowID) (*WorkflowExecutionRow, error)

LockAndGetWorkflowExecution locks and loads a workflow_execution row without its payload. Use LockAndGetWorkflowExecutionWithPayload when the WorkflowExecutionPayload is needed.

type WorkflowExecutionRowWithPayload 

type WorkflowExecutionRowWithPayload struct {
	WorkflowExecutionRow
	Payload *WorkflowExecutionPayload
}

WorkflowExecutionRowWithPayload is a WorkflowExecutionRow whose WorkflowExecutionPayload was additionally decoded during the scan.

func LockAndGetWorkflowExecutionWithPayload 

func LockAndGetWorkflowExecutionWithPayload(ctx context.Context, tx pgx.Tx, shardID ShardID, id WorkflowID) (*WorkflowExecutionRowWithPayload, error)

LockAndGetWorkflowExecutionWithPayload is LockAndGetWorkflowExecution that also decodes the WorkflowExecutionPayload.

type WorkflowExecutionState 

type WorkflowExecutionState int16

WorkflowExecutionState represents the lifecycle state of a workflow execution. Important: do not change constants! They are used in DB data. 

const (
	WorkflowExecutionStateUnknown   WorkflowExecutionState = 0
	WorkflowExecutionStateRunning   WorkflowExecutionState = 1
	WorkflowExecutionStateCompleted WorkflowExecutionState = 2
	WorkflowExecutionStateFailed    WorkflowExecutionState = 3
	WorkflowExecutionStateCanceled  WorkflowExecutionState = 4
	WorkflowExecutionStateTimedOut  WorkflowExecutionState = 5
	// WorkflowExecutionStateSystemFailed is a terminal failure caused by the
	// system rather than the workflow's own code: the abort retry budget was
	// exhausted, or a deterministic internal error (proto decode, unknown
	// workflow/activity, build-context failure). Distinct from Failed, which
	// is a user/business failure returned by the workflow function.
	WorkflowExecutionStateSystemFailed WorkflowExecutionState = 6
)

func (WorkflowExecutionState) IsTerminal 

func (s WorkflowExecutionState) IsTerminal() bool

func (WorkflowExecutionState) String 

func (s WorkflowExecutionState) String() string

type WorkflowID 

type WorkflowID uuid.UUID //nolint: recvcheck

WorkflowID identifies a workflow execution.

func NewWorkflowID added in v19.2.0 

func NewWorkflowID() WorkflowID

func (WorkflowID) Bytes 

func (id WorkflowID) Bytes() []byte

Bytes returns the UUID as a byte slice, for use in protobuf fields.

func (*WorkflowID) ScanUUID 

func (id *WorkflowID) ScanUUID(v pgtype.UUID) error

func (WorkflowID) String 

func (id WorkflowID) String() string

func (WorkflowID) UUIDValue 

func (id WorkflowID) UUIDValue() (pgtype.UUID, error)

type WorkflowIdentityResult 

type WorkflowIdentityResult struct {
	WorkflowID WorkflowID
	ShardID    ShardID
}

WorkflowIdentityResult is returned by TryCreateWorkflowIdentity.

type WorkflowInfo 

type WorkflowInfo struct {
	ShardID     ShardID
	WorkflowID  WorkflowID
	Name        string
	IdentityKey string
	State       WorkflowExecutionState
	CreatedAt   time.Time
	UpdatedAt   time.Time
	Output      []byte
	Error       string
}

WorkflowInfo is the read model returned by GetWorkflowInfo: a read-only, curated projection of a workflow_execution row together with its terminal result. Output holds the workflow's marshaled output and is set only when State is Completed; Error holds the failure message and is set only when State is Failed. Both are empty while the workflow is still running, or when it was canceled or timed out.

func GetWorkflowInfo 

func GetWorkflowInfo(ctx context.Context, db DB, shardID ShardID, id WorkflowID) (*WorkflowInfo, error)

GetWorkflowInfo returns a non-locking snapshot of a workflow's execution state together with its terminal result, in a single query. Unlike the LockAndGet* readers it takes no FOR UPDATE lock: it serves external status reads, not the per-workflow write path.

Once a workflow reaches a terminal state no further events are appended, and the terminal event is committed atomically with the terminal state, so a single unfenced read observes a consistent (state, result) pair. Returns ErrWorkflowNotFound when the workflow execution does not exist.

type WorkflowTaskID 

type WorkflowTaskID uuid.UUID //nolint: recvcheck

WorkflowTaskID identifies a workflow task within a workflow.

func NewWorkflowTaskID 

func NewWorkflowTaskID() WorkflowTaskID

func (WorkflowTaskID) Bytes 

func (id WorkflowTaskID) Bytes() []byte

Bytes returns the UUID as a byte slice, for use in protobuf fields.

func (*WorkflowTaskID) ScanUUID 

func (id *WorkflowTaskID) ScanUUID(v pgtype.UUID) error

func (WorkflowTaskID) String 

func (id WorkflowTaskID) String() string

func (WorkflowTaskID) UUIDValue 

func (id WorkflowTaskID) UUIDValue() (pgtype.UUID, error)

type WorkflowTaskInsert 

type WorkflowTaskInsert struct {
	ShardID           ShardID
	WorkflowID        WorkflowID
	WorkflowTaskID    WorkflowTaskID
	Payload           []byte
	State             WorkflowTaskState
	WorkflowCreatedAt time.Time
}

WorkflowTaskInsert holds the columns written when creating a workflow_task.

type WorkflowTaskPayload 

type WorkflowTaskPayload struct {
	Observability *WorkflowTaskPayload_Observability `protobuf:"bytes,1,opt,name=observability" json:"observability,omitempty"`
	// contains filtered or unexported fields
}

func (*WorkflowTaskPayload) Descriptor deprecated 

func (*WorkflowTaskPayload) Descriptor() ([]byte, []int)

Deprecated: Use WorkflowTaskPayload.ProtoReflect.Descriptor instead.

func (*WorkflowTaskPayload) GetObservability 

func (x *WorkflowTaskPayload) GetObservability() *WorkflowTaskPayload_Observability

func (*WorkflowTaskPayload) ProtoMessage 

func (*WorkflowTaskPayload) ProtoMessage()

func (*WorkflowTaskPayload) ProtoReflect 

func (x *WorkflowTaskPayload) ProtoReflect() protoreflect.Message

func (*WorkflowTaskPayload) Reset 

func (x *WorkflowTaskPayload) Reset()

func (*WorkflowTaskPayload) ScanBytes 

func (x *WorkflowTaskPayload) ScanBytes(src []byte) error

func (*WorkflowTaskPayload) String 

func (x *WorkflowTaskPayload) String() string

type WorkflowTaskPayload_Observability 

type WorkflowTaskPayload_Observability struct {
	Parent *SpanContext `protobuf:"bytes,1,opt,name=parent" json:"parent,omitempty"`
	// contains filtered or unexported fields
}

func (*WorkflowTaskPayload_Observability) Descriptor deprecated 

func (*WorkflowTaskPayload_Observability) Descriptor() ([]byte, []int)

Deprecated: Use WorkflowTaskPayload_Observability.ProtoReflect.Descriptor instead.

func (*WorkflowTaskPayload_Observability) GetParent 

func (x *WorkflowTaskPayload_Observability) GetParent() *SpanContext

func (*WorkflowTaskPayload_Observability) ProtoMessage 

func (*WorkflowTaskPayload_Observability) ProtoMessage()

func (*WorkflowTaskPayload_Observability) ProtoReflect 

func (x *WorkflowTaskPayload_Observability) ProtoReflect() protoreflect.Message

func (*WorkflowTaskPayload_Observability) Reset 

func (x *WorkflowTaskPayload_Observability) Reset()

func (*WorkflowTaskPayload_Observability) String 

func (x *WorkflowTaskPayload_Observability) String() string

type WorkflowTaskRow 

type WorkflowTaskRow struct {
	ShardID           ShardID
	WorkflowID        WorkflowID
	WorkflowTaskID    WorkflowTaskID
	Payload           *WorkflowTaskPayload
	State             WorkflowTaskState
	AbortRetryAttempt int32
	CreatedAt         time.Time
	ClaimedAt         *time.Time
	WorkflowCreatedAt time.Time
}

WorkflowTaskRow is the read model scanned from workflow_task. Its Payload is decoded into a WorkflowTaskPayload during the scan.

func BulkClaimWorkflowTasks 

func BulkClaimWorkflowTasks(ctx context.Context, tx pgx.Tx, shardIDs []ShardID, totalLimit int, excludeWorkflowIDs []WorkflowID) ([]WorkflowTaskRow, error)

type WorkflowTaskState 

type WorkflowTaskState int16

WorkflowTaskState represents the lifecycle state of a workflow task. Important: do not change constants! They are used in DB data. 

const (
	WorkflowTaskStateUnknown  WorkflowTaskState = 0
	WorkflowTaskStatePending  WorkflowTaskState = 1
	WorkflowTaskStateRunning  WorkflowTaskState = 2
	WorkflowTaskStateRetrying WorkflowTaskState = 3
)

func (WorkflowTaskState) String 

func (s WorkflowTaskState) String() string

type WorkloadStore 

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

WorkloadStore owns workload queries: arbitrary transactions (Tx), fenced shard-scoped transactions (WithSharedShardFence), and ambient DB access. It wraps the workload pgxpool.Pool.

func NewWorkloadStore 

func NewWorkloadStore(pool *pgxpool.Pool) *WorkloadStore

func (*WorkloadStore) DB 

func (s *WorkloadStore) DB() DB

func (*WorkloadStore) WithMultiShardFence 

func (s *WorkloadStore) WithMultiShardFence(
	ctx context.Context,
	shardIDs []ShardID,
	fenceIDs []FenceID,
	fn func(tx pgx.Tx, validShardIDs []ShardID) error,
) error

WithMultiShardFence runs fn inside a transaction that holds FOR KEY SHARE on every shard row whose (shard_id, fence_id) matches the caller's expected ownership. fn receives the validated shard IDs; shards whose live fence_id has moved on are silently filtered out. Partial fencing is the normal case for cross-shard operations, so this does NOT return ErrFenced — the caller scopes its own work to validShardIDs.

shardIDs and fenceIDs are parallel slices: index i carries the expected fence_id for shardIDs[i]. The caller is expected to maintain these in pre-allocated form (rebuilt only on shard ownership changes) so the per-call cost is one slice-header copy rather than a per-tick map-to-slice reallocation. Length mismatch is a programmer error and panics; nil/empty slices are valid (fn runs with no validated shards and no fence query).

This is the multi-shard analog of WithSharedShardFence. Lock-class semantics, conflict matrix, and rationale are documented on WithSharedShardFence and on multiShardFenceCheck in shard.go.

Cost: one extra round-trip vs. inlining the multiShardFenceCheck CTE into the caller's statement. Negligible at the bulk-claim poll rate. If a future caller is hot enough that this matters, inline the CTE in the caller's SQL rather than reaching for this helper.

func (*WorkloadStore) WithSharedShardFence 

func (s *WorkloadStore) WithSharedShardFence(ctx context.Context, shardID ShardID, fenceID FenceID, fn func(pgx.Tx) error) error

WithSharedShardFence runs fn in a transaction that holds a FOR KEY SHARE lock on the shard row, verifying fence_id matches the expected epoch. Multiple calls on the same shard can execute concurrently (FOR KEY SHARE is compatible with itself); fence bumps (ClaimShards) block behind any in-flight call because their explicit FOR UPDATE conflicts with FOR KEY SHARE.

FOR KEY SHARE is chosen over FOR SHARE because it is the weakest row lock that still blocks FOR UPDATE, and — crucially — it does NOT conflict with the FOR NO KEY UPDATE taken by the shard heartbeat UPDATE. That lets heartbeats proceed concurrently with in-flight fenced transactions on the same row, which matters under high per-shard concurrency. See the comment on fenceCheck in shard.go for the full lock-class rationale and invariants.

Per-workflow writes on the same workflow still serialize via FOR UPDATE on the workflow_execution row inside LockAndGetWorkflowExecution. The revision OCC check on UpdateWorkflowExecution remains as a latent safety net and is unreachable when callers hold the FOR UPDATE anchor from LockAndGetWorkflowExecution (see UpdateWorkflowExecution's contract). Callers must use LockAndGetWorkflowExecution (or equivalent per-row locking) for any write that requires per-workflow linearizability.

We don't use pgx.ReadOnly because FOR KEY SHARE is a locking operation that PostgreSQL disallows in read-only transactions.

Source Files

View all Source files

Directories

Path Synopsis

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.