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

workflow

package
v0.2.48 Latest Latest
Warning

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

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

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrCursor = errors.New("event cursor expired; refetch current state")
View Source 
var ErrFence = errors.New("lease or fencing token is no longer valid")
View Source 
var ErrNoWork = errors.New("no eligible work")
View Source 
var ErrPaused = errors.New("workflow scope is paused")
View Source 
var ErrPolicy = errors.New("policy changed or does not permit this action")
View Source 
var ErrReconciliation = errors.New("external outcome must be reconciled before retry")

Functions

func EmitTx 

func EmitTx(ctx context.Context, tx pgx.Tx, event domain.Event) error

Types

type Completion 

type Completion struct {
	Outcome      string `json:"outcome"`
	Retryable    bool   `json:"retryable"`
	Reason       string `json:"reason,omitempty"`
	RetryAfterMS int64  `json:"retry_after_ms,omitempty"`
}

type EnqueueInput 

type EnqueueInput struct {
	RepositoryID      string `json:"repository_id"`
	Recipe            string `json:"recipe"`
	RecipeVersion     string `json:"recipe_version"`
	TargetBranch      string `json:"target_branch"`
	ModelConnectionID string `json:"model_connection_id,omitempty"`
	ModelRoute        string `json:"model_route"`
	CampaignID        string `json:"campaign_id,omitempty"`
	RunnerPoolID      string `json:"runner_pool_id,omitempty"`
	PolicyHash        string `json:"policy_hash"`
	IdempotencyKey    string `json:"idempotency_key"`
	MaxAttempts       int    `json:"max_attempts"`
	Priority          int    `json:"priority"`
}

type EventPage 

type EventPage struct {
	ScanAfter int64          `json:"-"`
	Items     []domain.Event `json:"items"`
	Cursor    int64          `json:"cursor"`
	Complete  bool           `json:"complete"`
}

type Intent 

type Intent struct {
	ID            string          `json:"id"`
	OrgID         string          `json:"org_id"`
	TaskID        string          `json:"task_id"`
	RepositoryID  string          `json:"repository_id"`
	OperationID   string          `json:"operation_id"`
	Kind          string          `json:"kind"`
	State         string          `json:"state"`
	Payload       json.RawMessage `json:"payload"`
	DispatchCount int             `json:"dispatch_count"`
	Evidence      string          `json:"evidence"`
}

type Lease 

type Lease struct {
	OrgID        string    `json:"org_id"`
	RepositoryID string    `json:"repository_id"`
	TaskID       string    `json:"task_id"`
	JobID        string    `json:"job_id"`
	AttemptID    string    `json:"attempt_id"`
	OperationID  string    `json:"operation_id"`
	WorkerID     string    `json:"worker_id"`
	Fence        int64     `json:"fence"`
	PolicyHash   string    `json:"policy_hash"`
	ExpiresAt    time.Time `json:"expires_at"`
}

type Pause 

type Pause struct {
	Kind    string `json:"kind"`
	ID      string `json:"id"`
	Paused  bool   `json:"paused"`
	Version int64  `json:"version"`
}

type PolicyCheck 

type PolicyCheck func(context.Context, pgx.Tx, Task, string) (string, error)

type ScopeCheck 

type ScopeCheck func(context.Context, pgx.Tx, string, string, string) error

type Service 

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

func New 

func New(db *store.Store, identity *auth.Service, policy PolicyCheck) *Service

func (*Service) Advance 

func (s *Service) Advance(ctx context.Context, l Lease, next domain.TaskState) (Task, error)

func (*Service) AdvanceTx 

func (s *Service) AdvanceTx(ctx context.Context, tx pgx.Tx, l Lease, next domain.TaskState) (Task, error)

func (*Service) BeginIntent 

func (s *Service) BeginIntent(ctx context.Context, l Lease, id string) (Intent, error)

func (*Service) BeginIntentTx 

func (s *Service) BeginIntentTx(ctx context.Context, tx pgx.Tx, l Lease, id string) (Intent, error)

func (*Service) Cancel 

func (s *Service) Cancel(ctx context.Context, session auth.Session, orgID, id string, expected int64, requestID string) (Task, error)

func (*Service) CheckProposalTx 

func (s *Service) CheckProposalTx(ctx context.Context, tx pgx.Tx, t Task) (string, error)

func (*Service) Claim 

func (s *Service) Claim(ctx context.Context, workerID string, allowedOrgIDs []string, ttl time.Duration) (Lease, error)

func (*Service) ClaimScoped 

func (s *Service) ClaimScoped(ctx context.Context, workerID, orgID, poolID string, repositories []string, ttl time.Duration) (Lease, error)

func (*Service) ClaimScopedTx 

func (s *Service) ClaimScopedTx(ctx context.Context, tx pgx.Tx, workerID, orgID, poolID string, repositories []string, ttl time.Duration) (Lease, error)

func (*Service) Complete 

func (s *Service) Complete(ctx context.Context, l Lease, result Completion) (Task, error)

func (*Service) CompleteIntent 

func (s *Service) CompleteIntent(ctx context.Context, l Lease, id, outcome, evidence string) (Intent, error)

func (*Service) CompleteTx 

func (s *Service) CompleteTx(ctx context.Context, tx pgx.Tx, l Lease, result Completion) (Task, error)

func (*Service) ConcludeReconciledTx 

func (s *Service) ConcludeReconciledTx(ctx context.Context, tx pgx.Tx, orgID, taskID, actor, request string) error

func (*Service) Enqueue 

func (s *Service) Enqueue(ctx context.Context, session auth.Session, orgID string, in EnqueueInput, requestID string) (Task, error)

func (*Service) EnqueuePrepared 

func (s *Service) EnqueuePrepared(ctx context.Context, session auth.Session, orgID string, in EnqueueInput, requestID string, prepare func(context.Context, pgx.Tx, Task) error) (Task, error)

func (*Service) Get 

func (s *Service) Get(ctx context.Context, session auth.Session, orgID, id string) (Task, error)

func (*Service) Heartbeat 

func (s *Service) Heartbeat(ctx context.Context, l Lease, ttl time.Duration) (Lease, error)

func (*Service) InvalidateJobTx 

func (s *Service) InvalidateJobTx(ctx context.Context, tx pgx.Tx, orgID, workerID, jobID string) error

func (*Service) InvalidateWorkerTx 

func (s *Service) InvalidateWorkerTx(ctx context.Context, tx pgx.Tx, orgID, workerID string) error

func (*Service) List 

func (s *Service) List(ctx context.Context, session auth.Session, orgID, state string, limit int, cursor string) (domain.Page[Task], error)

func (*Service) PausedTx 

func (s *Service) PausedTx(ctx context.Context, tx pgx.Tx, orgID, id string) (bool, error)

func (*Service) PrepareIntentTx 

func (s *Service) PrepareIntentTx(ctx context.Context, tx pgx.Tx, l Lease, kind, key string, payload json.RawMessage) (Intent, error)

func (*Service) PruneEvents 

func (s *Service) PruneEvents(ctx context.Context, orgID string, through int64) error

func (*Service) ReconcileIntent 

func (s *Service) ReconcileIntent(ctx context.Context, orgID, id, outcome, evidence string) (Intent, error)

func (*Service) ReconcileIntentTx 

func (s *Service) ReconcileIntentTx(ctx context.Context, tx pgx.Tx, orgID, taskID, id, outcome, evidence string) (Intent, error)

func (*Service) Recover 

func (s *Service) Recover(ctx context.Context, orgID string) error

func (*Service) RegisterCampaignAuthority 

func (s *Service) RegisterCampaignAuthority(check func(context.Context, pgx.Tx, Task) error)

func (*Service) RegisterScopeCheck 

func (s *Service) RegisterScopeCheck(check ScopeCheck)

func (*Service) Replay 

func (s *Service) Replay(ctx context.Context, session auth.Session, orgID string, after int64, limit int) (EventPage, error)

func (*Service) Resume 

func (s *Service) Resume(ctx context.Context, session auth.Session, orgID, id string, expected int64, requestID string) (Task, error)

func (*Service) SetCampaignPauseTx 

func (s *Service) SetCampaignPauseTx(ctx context.Context, tx pgx.Tx, org, id string, paused bool, actor, request string) error

func (*Service) SetPause 

func (s *Service) SetPause(ctx context.Context, session auth.Session, orgID string, p Pause, expected int64, requestID string) (Pause, error)

func (*Service) ValidateFenceTx 

func (s *Service) ValidateFenceTx(ctx context.Context, tx pgx.Tx, l Lease, action string) (Task, error)

type Task 

type Task struct {
	ID                    string           `json:"id"`
	OrgID                 string           `json:"org_id"`
	RepositoryID          string           `json:"repository_id"`
	OperationID           string           `json:"operation_id"`
	Recipe                string           `json:"recipe"`
	RecipeVersion         string           `json:"recipe_version"`
	TargetBranch          string           `json:"target_branch"`
	ModelConnectionID     string           `json:"model_connection_id,omitempty"`
	ModelRoute            string           `json:"model_route"`
	CampaignID            string           `json:"campaign_id,omitempty"`
	RunnerPoolID          string           `json:"runner_pool_id,omitempty"`
	PolicyHash            string           `json:"policy_hash"`
	StartingPolicyHash    string           `json:"starting_policy_hash"`
	State                 domain.TaskState `json:"state"`
	Reason                string           `json:"reason"`
	Version               int64            `json:"version"`
	CancelVersion         int64            `json:"cancel_version"`
	CancellationRequested bool             `json:"cancellation_requested"`
	MaxAttempts           int              `json:"max_attempts"`
	CreatedAt             time.Time        `json:"created_at"`
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.