activity

package
v2.8.1 Latest Latest
Warning

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

 Go to latest Published: Aug 19, 2026 License: BSD-3-Clause Imports: 32 Imported by: 0

Documentation

Overview

Package activity owns background activity persistence, execution tracking, streaming, and the HTTP surface used to inspect and cancel work.

Index

Constants

View Source 
const ErrActivityNotCancelable = errors.Sentinel("activity is not cancelable")

ErrActivityNotCancelable indicates the activity has already reached a terminal state and can no longer be cancelled.

Variables

This section is empty.

Functions

func RegisterActivities 

func RegisterActivities(api huma.API, h *ActivityHandler)

Types

type Activity added in v2.8.1 

type Activity struct {
	database.BaseModel

	EnvironmentID        string               `json:"environmentId" gorm:"column:environment_id;not null;index" sortable:"true"`
	BatchID              *string              `json:"batchId,omitempty" gorm:"column:batch_id;index"`
	Type                 activitytypes.Type   `json:"type" gorm:"column:type;not null;index" sortable:"true"`
	Status               activitytypes.Status `json:"status" gorm:"column:status;not null;index" sortable:"true"`
	ResourceType         *string              `json:"resourceType,omitempty" gorm:"column:resource_type;index" sortable:"true"`
	ResourceID           *string              `json:"resourceId,omitempty" gorm:"column:resource_id;index"`
	ResourceName         *string              `json:"resourceName,omitempty" gorm:"column:resource_name" sortable:"true"`
	Progress             *int                 `json:"progress,omitempty" gorm:"column:progress"`
	Step                 string               `json:"step,omitempty" gorm:"column:step"`
	LatestMessage        string               `json:"latestMessage,omitempty" gorm:"column:latest_message"`
	StartedByUserID      *string              `json:"startedByUserId,omitempty" gorm:"column:started_by_user_id;index"`
	StartedByUsername    *string              `json:"startedByUsername,omitempty" gorm:"column:started_by_username"`
	StartedByDisplayName *string              `json:"startedByDisplayName,omitempty" gorm:"column:started_by_display_name"`
	StartedAt            time.Time            `json:"startedAt" gorm:"column:started_at;not null" sortable:"true"`
	EndedAt              *time.Time           `json:"endedAt,omitempty" gorm:"column:ended_at" sortable:"true"`
	DurationMs           *int64               `json:"durationMs,omitempty" gorm:"column:duration_ms" sortable:"true"`
	Error                *string              `json:"error,omitempty" gorm:"column:error"`
	Metadata             database.JSON        `json:"metadata,omitempty" gorm:"type:text"`
}

func (Activity) TableName added in v2.8.1 

func (Activity) TableName() string

type ActivityHandler 

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

func NewHandler 

func NewHandler(activityService *ActivityService, environment EnvironmentDependencies) *ActivityHandler

func (*ActivityHandler) CancelActivity 

func (h *ActivityHandler) CancelActivity(ctx context.Context, input *CancelActivityInput) (*CancelActivityOutput, error)

func (*ActivityHandler) ClearHistory 

func (h *ActivityHandler) ClearHistory(ctx context.Context, input *ClearActivityHistoryInput) (*ClearActivityHistoryOutput, error)

func (*ActivityHandler) GetActivity 

func (h *ActivityHandler) GetActivity(ctx context.Context, input *GetActivityInput) (*GetActivityOutput, error)

func (*ActivityHandler) ListActivities 

func (h *ActivityHandler) ListActivities(ctx context.Context, input *ListActivitiesInput) (*ListActivitiesOutput, error)

func (*ActivityHandler) RunLocalStreamProducer 

func (h *ActivityHandler) RunLocalStreamProducer(ctx context.Context, limit int, events chan<- activitytypes.StreamEvent)

func (*ActivityHandler) RunRemoteStreamPollers 

func (h *ActivityHandler) RunRemoteStreamPollers(ctx context.Context, ps *authz.PermissionSet, limit int, events chan<- activitytypes.StreamEvent)

RunRemoteStreamPollers keeps one poller goroutine per enabled remote environment, re-listing periodically so environments added or removed while the stream is open are picked up without a reconnect.

type ActivityMessage added in v2.8.1 

type ActivityMessage struct {
	database.BaseModel

	ActivityID string                     `json:"activityId" gorm:"column:activity_id;not null;index"`
	Level      activitytypes.MessageLevel `json:"level" gorm:"column:level;not null"`
	Message    string                     `json:"message" gorm:"column:message;not null"`
	Payload    database.JSON              `json:"payload,omitempty" gorm:"type:text"`
	Activity   *Activity                  `json:"-" gorm:"foreignKey:ActivityID;constraint:OnDelete:CASCADE"`
}

func (ActivityMessage) TableName added in v2.8.1 

func (ActivityMessage) TableName() string

type ActivityService 

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

func NewActivityService 

func NewActivityService(db *database.DB, settingsService *settings.SettingsService) *ActivityService

func (*ActivityService) AppendMessage 

func (s *ActivityService) AppendMessage(ctx context.Context, activityID string, req AppendActivityMessageRequest) (*activitytypes.Message, error)

func (*ActivityService) AppendMessages 

func (s *ActivityService) AppendMessages(ctx context.Context, activityID string, reqs []AppendActivityMessageRequest) ([]activitytypes.Message, error)

AppendMessages persists a batch of output lines in one transaction — a single multi-row message INSERT plus one coalesced Activity update — instead of an INSERT+UPDATE+re-SELECT transaction per line, which turned an image pull into hundreds of fsync'd SQLite transactions. The activity publish DTO is re-SELECTed inside the transaction after the update (once per batch, not per line) so it reflects lifecycle fields a concurrent CompleteActivity/UpdateActivity committed first; a terminal write that commits after this transaction but publishes before this snapshot is handled by admitActivityPublishInternal dropping the stale event.

func (*ActivityService) AwaitActivitySlot 

func (s *ActivityService) AwaitActivitySlot(ctx context.Context, activityID, environmentID string) error

AwaitActivitySlot blocks until the queued activity holds a concurrency slot, then flips its status to running. It returns immediately when the activity already took a slot at creation. On cancellation the context cause is returned and the activity stays queued for its caller to finalize. Implements activitylib.SlotWaiter.

func (*ActivityService) AwaitActivitySlotBounded 

func (s *ActivityService) AwaitActivitySlotBounded(ctx context.Context, activityID, environmentID string) error

AwaitActivitySlotBounded waits for a concurrency slot like AwaitActivitySlot but gives up after timeouts.DefaultActivitySlotWait, returning ActivitySlotWaitTimeoutError so the caller fails the queued activity loudly instead of parking forever behind long-running slot holders.

func (*ActivityService) CancelActivity 

func (s *ActivityService) CancelActivity(ctx context.Context, environmentID, activityID, requestedBy string) (*activitytypes.Activity, error)

CancelActivity requests cancellation of a running or queued activity. When the activity's work is running in this process it interrupts it (the work finalizes its own terminal status); otherwise it marks the activity cancelled directly, but only if it is still active. Returns ErrActivityNotCancelable if the activity has already reached a terminal state, or gorm.ErrRecordNotFound if it is unknown.

func (*ActivityService) CompleteActivity 

func (s *ActivityService) CompleteActivity(ctx context.Context, activityID string, status activitytypes.Status, finalMessage string, errMessage *string, finalStep ...string) (*activitytypes.Activity, error)

func (*ActivityService) DeleteHistory 

func (s *ActivityService) DeleteHistory(ctx context.Context, environmentID string) (int64, error)

func (*ActivityService) FailAbandonedActivities 

func (s *ActivityService) FailAbandonedActivities(ctx context.Context) (int64, error)

FailAbandonedActivities marks queued/running activities whose worker is no longer alive in this process as failed, releasing any concurrency slot they still hold. Liveness comes from the running map, not age: every creation path Tracks its activity right after StartActivity, and the short grace period covers that create→Track window. This assumes exactly one Arcane process owns the database (managers and agents each own theirs); running multiple replicas against one database would make each replica sweep the other's live work.

The terminal write is status-guarded like CancelActivity's untracked fallback: a worker completing concurrently wins the race and the row is skipped here.

func (*ActivityService) FailStaleImageUpdateChecks 

func (s *ActivityService) FailStaleImageUpdateChecks(ctx context.Context) (int64, error)

FailStaleImageUpdateChecks marks image update checks that were left running across a prior process lifetime as failed. It intentionally scopes cleanup to old image-update-check activities so startup repair cannot affect other work.

func (*ActivityService) GetActivityDetail 

func (s *ActivityService) GetActivityDetail(ctx context.Context, environmentID, activityID string, limit int) (*activitytypes.Detail, error)

func (*ActivityService) ListActivitiesPaginated 

func (s *ActivityService) ListActivitiesPaginated(ctx context.Context, environmentID string, params pagination.QueryParams) ([]activitytypes.Activity, pagination.Response, error)

func (*ActivityService) PatchActivityMetadata 

func (s *ActivityService) PatchActivityMetadata(ctx context.Context, activityID string, patch database.JSON) error

PatchActivityMetadata merges patch into the activity's existing metadata, unlike UpdateActivity which replaces the metadata wholesale.

func (*ActivityService) PruneHistory 

func (s *ActivityService) PruneHistory(ctx context.Context, retentionDays, maxEntries int) (int64, error)

func (*ActivityService) RequestCancel 

func (s *ActivityService) RequestCancel(activityID string) bool

RequestCancel cancels the work context registered for activityID, signalling activitylib.ErrCanceled as the cause. It returns whether a running activity was found in this process.

func (*ActivityService) ResolveOrphanedQueuedActivities 

func (s *ActivityService) ResolveOrphanedQueuedActivities(ctx context.Context) (int64, error)

ResolveOrphanedQueuedActivities fails any activity still queued at startup. Queued state is owned by a live goroutine blocked on AwaitActivitySlot, so a queued row after a restart can never start running.

func (*ActivityService) ResolveStaleAutoUpdateActivities 

func (s *ActivityService) ResolveStaleAutoUpdateActivities(ctx context.Context) (int64, error)

ResolveStaleAutoUpdateActivities finalizes auto-update activities left running by a prior process lifetime. A run whose metadata marks a triggered self-update completed by restarting Arcane, so it is recorded as success; anything else still running at startup was interrupted and is failed.

func (*ActivityService) StartActivity 

func (s *ActivityService) StartActivity(ctx context.Context, req StartActivityRequest) (*activitytypes.Activity, error)

func (*ActivityService) Subscribe 

func (s *ActivityService) Subscribe(environmentID string) (<-chan activitytypes.StreamEvent, func() bool, func())

func (*ActivityService) Track 

func (s *ActivityService) Track(ctx context.Context, activityID string) context.Context

Track derives a cancelable work context bound to activityID and registers its cancel function so RequestCancel can interrupt the work. The registration is released when the activity is completed (see CompleteActivity) or when the returned context is otherwise no longer needed. Implements activitylib.Tracker.

func (*ActivityService) UpdateActivity 

func (s *ActivityService) UpdateActivity(ctx context.Context, activityID string, req UpdateActivityRequest) (*activitytypes.Activity, error)

type AppendActivityMessageRequest 

type AppendActivityMessageRequest = activitylib.AppendMessageRequest

type CancelActivityInput 

type CancelActivityInput struct {
	EnvironmentID string `path:"id" doc:"Environment ID"`
	ActivityID    string `path:"activityId" doc:"Activity ID"`
	RequestedBy   string `query:"requestedBy" doc:"Display name to attribute the cancellation to (used when proxying to a remote environment)"`
}

type CancelActivityOutput 

type CancelActivityOutput struct {
	Body base.ApiResponse[activitytypes.Activity]
}

type ClearActivityHistoryInput 

type ClearActivityHistoryInput struct {
	EnvironmentID string `path:"id" doc:"Environment ID"`
}

type ClearActivityHistoryOutput 

type ClearActivityHistoryOutput struct {
	Body base.ApiResponse[activitytypes.ClearHistoryResult]
}

type Dependencies 

type Dependencies struct {
	DB          *database.DB
	Settings    *settings.SettingsService
	Environment EnvironmentDependencies
}

type EnvironmentDependencies 

type EnvironmentDependencies struct {
	ProxyJSONRequest               handlerutil.RemoteJSONProxy
	ListRemoteEnvironments         func(context.Context) ([]environment.Environment, error)
	GetActiveRemoteEnvironment     func(string) mo.Option[environment.Environment]
	ProxyJSONRequestForEnvironment func(context.Context, environment.Environment, string, string, []byte, any) error
	ResolveEnvironmentName         func(context.Context, string) string
}

type GetActivityInput 

type GetActivityInput struct {
	EnvironmentID string `path:"id" doc:"Environment ID"`
	ActivityID    string `path:"activityId" doc:"Activity ID"`
	Limit         int    `query:"limit" default:"500" doc:"Maximum messages to return"`
}

type GetActivityOutput 

type GetActivityOutput struct {
	Body base.ApiResponse[activitytypes.Detail]
}

type ListActivitiesInput 

type ListActivitiesInput struct {
	EnvironmentID string `path:"id" doc:"Environment ID"`
	Search        string `query:"search" doc:"Search query"`
	Sort          string `query:"sort" doc:"Column to sort by"`
	Order         string `query:"order" default:"desc" doc:"Sort direction"`
	Start         int    `query:"start" default:"0" doc:"Start index"`
	Limit         int    `query:"limit" default:"50" doc:"Limit"`
	Status        string `query:"status" doc:"Filter by activity status"`
	Type          string `query:"type" doc:"Filter by activity type"`
	ResourceType  string `query:"resourceType" doc:"Filter by resource type"`
}

type ListActivitiesOutput 

type ListActivitiesOutput struct {
	Body base.Paginated[activitytypes.Activity]
}

type Module 

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

func New 

func New(deps Dependencies) *Module

func (*Module) Handler 

func (m *Module) Handler() *ActivityHandler

func (*Module) RegisterRoutes 

func (m *Module) RegisterRoutes(api huma.API)

func (*Module) Service 

func (m *Module) Service() *ActivityService

type StartActivityRequest 

type StartActivityRequest = activitylib.StartRequest

type StreamAllActivitiesInput 

type StreamAllActivitiesInput struct {
	Limit int `query:"limit" default:"50" doc:"Snapshot limit per environment"`
}

type UpdateActivityRequest 

type UpdateActivityRequest = activitylib.UpdateRequest

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.