repository

package
v0.0.0-...-fc2aea3 Latest Latest
Warning

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

 Go to latest Published: Nov 12, 2026 License: AGPL-3.0 Imports: 56 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ScheduledJobsChannel = "jobs"
)

Variables

View Source 
var (
	ErrEventReadUnsupported  = errors.New("event read unsupported")
	ErrEventWriteUnsupported = errors.New("event write unsupported")
)
View Source 
var ErrEvictionVictimsChanged = errors.New("eviction victims changed before placement committed")

ErrEvictionVictimsChanged means the chosen victims changed before the placement committed.

View Source 
var ErrInsufficientEvictableCapacity = errors.New("unable to schedule container, worker out of free and evictable capacity")
View Source 
var ErrInvalidTaskInterval = errors.New("invalid task aggregation interval")
View Source 
var ErrReplicaProtectionChanged = errors.New("replica protection assignment changed; retry reconciliation")

ErrReplicaProtectionChanged is retryable after the controller refreshes its replica snapshot. A stopping, evicted, or unassigned container cannot be promoted.

View Source 
var ErrWorkerCapacityChanged = errors.New("worker capacity changed after selection")

ErrWorkerCapacityChanged means the scheduler selected the worker from a stale snapshot.

View Source 
var PostgresDataError = pq.ErrorClass("22")

Functions

func AuthorizeImage 

func AuthorizeImage(ctx context.Context, backendRepo BackendRepository, workspaceID, imageID string) error

AuthorizeImage checks that a workspace may run imageID. A filesystem snapshot belongs to the workspace of the stub it was taken from. A built image is named by its build, which only that workspace can reproduce when the build used private inputs, so knowing its ID is enough.

func ContextWithVMDiskBurst 

func ContextWithVMDiskBurst(ctx context.Context) context.Context

ContextWithVMDiskBurst limits disk batching to VM preparation. Ordinary disk operations and existing-disk validation/resize do not take this path.

func DeliverWebhook 

func DeliverWebhook(ctx context.Context, webhook types.WorkspaceWebhook, event cloudevents.Event) error

DeliverWebhook posts one signed CloudEvent; non-2xx is an error.

func GenerateDSN 

func GenerateDSN(config types.PostgresConfig) string

func GenerateSubdomain 

func GenerateSubdomain(deploymentName, stubType string, workspaceId uint) string

GenerateSubdomain is deterministic, so a deployment's host URL is known before it exists.

func NewCacheRedisRepository 

func NewCacheRedisRepository(rdb *common.RedisClient) cache.CoordinatorRepository

func NewRedisClientForTest 

func NewRedisClientForTest() (*common.RedisClient, error)

func PrepareIdleWorkerRolloutContainer 

func PrepareIdleWorkerRolloutContainer(ctx context.Context, rdb *common.RedisClient, workerID, generation string, container types.ContainerState, stub *types.StubWithRelated) (bool, error)

PrepareIdleWorkerRolloutContainer fences request admission before marking an idle serverless container STOPPING. Persistent VMs, pods, and active jobs keep running; Scheduler.Stop performs the ordinary idempotent finalization.

Types

type AuthInfo 

type AuthInfo struct {
	Workspace *types.Workspace
	Token     *types.Token
}

type BackendRepository 

type BackendRepository interface {
	ListWorkspaces(ctx context.Context) ([]types.Workspace, error)
	CreateWorkspace(ctx context.Context) (types.Workspace, error)
	GetWorkspaceByExternalId(ctx context.Context, externalId string) (types.Workspace, error)
	GetWorkspaceByExternalIdWithSigningKey(ctx context.Context, externalId string) (types.Workspace, error)
	GetWorkspace(ctx context.Context, workspaceId uint) (*types.Workspace, error)
	GetWorkspaceStorage(ctx context.Context, storageId uint) (*types.WorkspaceStorage, error)
	CreateWorkspaceStorage(ctx context.Context, workspaceId uint, storage types.WorkspaceStorage) (*types.WorkspaceStorage, error)
	GetAdminWorkspace(ctx context.Context) (*types.Workspace, error)
	CreateObject(ctx context.Context, hash string, size int64, workspaceId uint) (*types.Object, error)
	GetObjectByHash(ctx context.Context, hash string, workspaceId uint) (*types.Object, error)
	GetObjectByExternalId(ctx context.Context, externalId string, workspaceId uint) (types.Object, error)
	GetObjectByExternalStubId(ctx context.Context, stubId string, workspaceId uint) (types.Object, error)
	UpdateObjectSizeByExternalId(ctx context.Context, externalId string, size int) error
	DeleteObjectByExternalId(ctx context.Context, externalId string) error
	CreateToken(ctx context.Context, workspaceId uint, tokenType string, reusable bool) (types.Token, error)
	CreateWorkerToken(ctx context.Context, workspaceId uint, tokenType string, workerId string) (types.Token, error)
	BindWorkerToken(ctx context.Context, tokenKey, workerId string) (bool, error)
	AuthorizeToken(ctx context.Context, tokenKey string) (*types.Token, *types.Workspace, error)
	RetrieveActiveToken(ctx context.Context, workspaceId uint, tokenType string) (*types.Token, error)
	GetTokenByExternalId(ctx context.Context, workspaceId uint, extTokenId string) (*types.Token, error)
	ListTokens(ctx context.Context, workspaceId uint) ([]types.Token, error)
	SetWorkspaceTokensDisabled(ctx context.Context, workspaceId uint, disabled bool) ([]string, error)
	ToggleToken(ctx context.Context, workspaceId uint, extTokenId string) (types.Token, error)
	DeleteToken(ctx context.Context, workspaceId uint, extTokenId string) error
	RevokeTokenByExternalId(ctx context.Context, externalId string) (string, error)
	GetTask(ctx context.Context, externalId string) (*types.Task, error)
	GetTaskStatus(ctx context.Context, externalId string) (types.TaskStatus, error)
	GetTaskWithRelated(ctx context.Context, externalId string) (*types.TaskWithRelated, error)
	GetTaskByWorkspace(ctx context.Context, externalId string, workspace *types.Workspace) (*types.TaskWithRelated, error)
	CreateTask(ctx context.Context, params *types.TaskParams) (*types.Task, error)
	UpdateTask(ctx context.Context, externalId string, updatedTask types.Task) (*types.Task, error)
	DeleteTask(ctx context.Context, externalId string) error
	ListTasks(ctx context.Context) ([]types.Task, error)
	ListTasksWithRelated(ctx context.Context, filters types.TaskFilter) ([]types.TaskWithRelated, error)
	ListTasksWithRelatedPaginated(ctx context.Context, filters types.TaskFilter) (common.CursorPaginationInfo[types.TaskWithRelated], error)
	AggregateTasksByTimeWindow(ctx context.Context, filters types.TaskFilter) ([]types.TaskCountByTime, error)
	GetTaskCountPerDeployment(ctx context.Context, filters types.TaskFilter) ([]types.TaskCountPerDeployment, error)
	GetOrCreateStub(ctx context.Context, name, stubType string, config types.StubConfigV1, objectId, workspaceId uint, forceCreate bool, appId uint) (types.Stub, error)
	UpdateStubConfig(ctx context.Context, stubId uint, config *types.StubConfigV1) error
	GetStubByExternalId(ctx context.Context, externalId string, queryFilters ...types.QueryFilter) (*types.StubWithRelated, error)
	GetStubById(ctx context.Context, stubId uint, queryFilters ...types.QueryFilter) (*types.StubWithRelated, error)
	GetDeploymentBySubdomain(ctx context.Context, subdomain string, version uint) (*types.DeploymentWithRelated, error)
	GetVolume(ctx context.Context, workspaceId uint, name string) (*types.Volume, error)
	GetVolumeByExternalId(ctx context.Context, workspaceId uint, externalId string) (*types.Volume, error)
	GetOrCreateVolume(ctx context.Context, workspaceId uint, name string) (*types.Volume, error)
	DeleteVolume(ctx context.Context, workspaceId uint, name string) error
	ListVolumesWithRelated(ctx context.Context, workspaceId uint) ([]types.VolumeWithRelated, error)
	ListDeploymentsWithRelated(ctx context.Context, filters types.DeploymentFilter) ([]types.DeploymentWithRelated, error)
	ListLatestDeploymentsByAppIDs(ctx context.Context, workspaceID uint, appExternalIDs []string) (map[string]types.DeploymentWithRelated, error)
	CountActiveDeploymentsByApp(ctx context.Context, workspaceID uint, appExternalIDs []string) (map[string]int, error)
	AggregateTaskActivityByApp(ctx context.Context, workspaceID uint, appExternalIDs []string, since time.Time) (map[string][]types.AppActivityBucket, error)
	ListLatestDeploymentsWithRelatedPaginated(ctx context.Context, filters types.DeploymentFilter) (common.CursorPaginationInfo[types.DeploymentWithRelated], error)
	ListDeploymentsPaginated(ctx context.Context, filters types.DeploymentFilter) (common.CursorPaginationInfo[types.DeploymentWithRelated], error)
	GetLatestDeploymentByName(ctx context.Context, workspaceId uint, name string, stubType string, filterDeleted bool) (*types.DeploymentWithRelated, error)
	GetDeploymentByExternalId(ctx context.Context, workspaceId uint, deploymentExternalId string) (*types.DeploymentWithRelated, error)
	GetAnyDeploymentByExternalId(ctx context.Context, deploymentExternalId string) (*types.DeploymentWithRelated, error)
	GetDeploymentByStubExternalId(ctx context.Context, workspaceId uint, stubExternalId string) (*types.DeploymentWithRelated, error)
	GetDeploymentByNameAndVersion(ctx context.Context, workspaceId uint, name string, version uint, stubType string) (*types.DeploymentWithRelated, error)
	CreateDeployment(ctx context.Context, workspaceId uint, name string, version uint, stubId uint, stubType string, appId uint) (*types.Deployment, error)
	UpdateDeployment(ctx context.Context, deployment types.Deployment) (*types.Deployment, error)
	DeleteDeployment(ctx context.Context, deployment types.Deployment) error
	ListStubs(ctx context.Context, filters types.StubFilter) ([]types.StubWithRelated, error)
	ListLatestStubsByAppIDs(ctx context.Context, workspaceID uint, appExternalIDs []string) (map[string]types.StubWithRelated, error)
	ListAppIDsByStubExternalIDs(ctx context.Context, workspaceID string, stubExternalIDs []string) (map[string]string, error)
	ListStubsPaginated(ctx context.Context, filters types.StubFilter) (common.CursorPaginationInfo[types.StubWithRelated], error)
	GetConcurrencyLimit(ctx context.Context, concurrenyLimitId uint) (*types.ConcurrencyLimit, error)
	GetConcurrencyLimitByWorkspaceId(ctx context.Context, workspaceId string) (*types.ConcurrencyLimit, error)
	DeleteConcurrencyLimit(ctx context.Context, workspaceId types.Workspace) error
	CreateConcurrencyLimit(ctx context.Context, workspaceId uint, gpuLimit uint32, cpuMillicoreLimit uint32) (*types.ConcurrencyLimit, error)
	UpdateConcurrencyLimit(ctx context.Context, concurrencyLimitId uint, gpuLimit uint32, cpuMillicoreLimit uint32) (*types.ConcurrencyLimit, error)
	ListConcurrencyLimitsByWorkspaceId(ctx context.Context, workspaceId string) ([]types.ConcurrencyLimit, error)
	RevertToConcurrencyLimit(ctx context.Context, workspaceId string, concurrencyLimitId string) (*types.ConcurrencyLimit, error)
	CreateSecret(ctx context.Context, workspace *types.Workspace, tokenId uint, name string, value string, validateName bool, hosts []string) (*types.Secret, error)
	GetSecretByName(ctx context.Context, workspace *types.Workspace, name string) (*types.Secret, error)
	GetSecretsByName(ctx context.Context, workspace *types.Workspace, names []string) ([]types.Secret, error)
	GetSecretByNameDecrypted(ctx context.Context, workspace *types.Workspace, name string) (*types.Secret, error)
	GetSecretsByNameDecrypted(ctx context.Context, workspace *types.Workspace, names []string) ([]types.Secret, error)
	ListSecrets(ctx context.Context, workspace *types.Workspace) ([]types.Secret, error)
	UpdateSecret(ctx context.Context, workspace *types.Workspace, tokenId uint, secretName string, value string, hosts []string) (*types.Secret, error)
	DeleteSecret(ctx context.Context, workspace *types.Workspace, secretName string) error
	ListStacks(ctx context.Context, workspaceId uint) ([]types.Stack, error)
	CreateStack(ctx context.Context, workspaceId uint, name string, spec json.RawMessage) (*types.Stack, error)
	UpdateStack(ctx context.Context, workspaceId uint, externalId, name string, spec json.RawMessage) (*types.Stack, error)
	DeleteStack(ctx context.Context, workspaceId uint, externalId string) error
	CreateScheduledJob(ctx context.Context, scheduledJob *types.ScheduledJob) (*types.ScheduledJob, error)
	DeleteScheduledJob(ctx context.Context, scheduledJob *types.ScheduledJob) error
	DeletePreviousScheduledJob(ctx context.Context, deployment *types.Deployment) error
	GetScheduledJob(ctx context.Context, deploymentId uint) (*types.ScheduledJob, error)
	ListenToChannel(ctx context.Context, channel string) (<-chan string, error)
	Ping() error
	GetTaskMetrics(ctx context.Context, periodStart time.Time, periodEnd time.Time) (types.TaskMetrics, error)
	GetOrCreateApp(ctx context.Context, workspaceId uint, appName string) (*types.App, error)
	RetrieveApp(ctx context.Context, workspaceId uint, appId string) (*types.App, error)
	RetrieveAppByStubExternalId(ctx context.Context, stubExternalId string) (*types.App, error)
	ListApps(ctx context.Context, workspaceId uint) ([]types.App, error)
	ListAppsPaginated(ctx context.Context, workspaceId uint, filters types.AppFilter) (common.CursorPaginationInfo[types.App], error)
	CountApps(ctx context.Context, workspaceId uint) (int, error)
	DeleteApp(ctx context.Context, appId string) error
	GetImageClipVersion(ctx context.Context, imageId string) (uint32, error)
	CreateImage(ctx context.Context, imageId string, clipVersion uint32) (uint32, error)
	SetImageCredentialSecretNames(ctx context.Context, imageId string, names []string) error
	GetImageCredentials(ctx context.Context, workspace *types.Workspace, imageId string) (string, error)
	CreateCheckpoint(ctx context.Context, checkpoint *types.Checkpoint, compatibilityKey string) (*types.Checkpoint, error)
	UpdateCheckpoint(ctx context.Context, checkpoint *types.Checkpoint) (*types.Checkpoint, error)
	ListCheckpoints(ctx context.Context, workspaceExternalId string) ([]types.Checkpoint, error)
	GetCheckpointById(ctx context.Context, checkpointId string) (*types.Checkpoint, error)
	GetLatestCheckpointByStubId(ctx context.Context, stubExternalId string, compatibilityKey ...string) (*types.Checkpoint, error)
	ListStaleCheckpoints(ctx context.Context, activeRecentStubKeys []string, stubLastUsedBefore time.Time) ([]types.Checkpoint, error)
	PruneCheckpoints(ctx context.Context, checkpointIds []string) ([]types.Checkpoint, error)
	CreateDiskSnapshot(ctx context.Context, snapshot *types.DiskSnapshot) (*types.DiskSnapshot, error)
	UpdateDiskSnapshot(ctx context.Context, snapshot *types.DiskSnapshot) (*types.DiskSnapshot, error)
	GetDiskSnapshot(ctx context.Context, workspaceId uint, snapshotId string) (*types.DiskSnapshot, error)
	PublishDiskSnapshot(ctx context.Context, workspaceId uint, snapshotId string) error
	GetLatestDiskSnapshot(ctx context.Context, workspaceId uint, diskName string) (*types.DiskSnapshot, error)
	ListDiskSnapshots(ctx context.Context, filter types.DiskSnapshotFilter) ([]types.DiskSnapshot, error)
	GetDisk(ctx context.Context, workspaceId uint, name string) (*types.Disk, error)
	GetOrCreateDisk(ctx context.Context, workspaceId uint, disk *types.Disk) (*types.Disk, error)
	// DeleteDisk unregisters the mutable disk resource. Immutable snapshots and
	// their backing payload are retained because snapshot IDs can outlive a disk
	// record (for example, when pinned by a template or published for a fork).
	DeleteDisk(ctx context.Context, workspaceId uint, name string) error
	ListDisksWithRelated(ctx context.Context, workspaceId uint) ([]types.DiskWithRelated, error)
}

func NewBackendPostgresRepositoryForTest 

func NewBackendPostgresRepositoryForTest() (BackendRepository, sqlmock.Sqlmock)

type CacheRedisRepository 

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

func (*CacheRedisRepository) CountCacheRegistrations 

func (r *CacheRedisRepository) CountCacheRegistrations(ctx context.Context, logicalHostID string) (int64, error)

func (*CacheRedisRepository) GetActiveCacheRegistration 

func (r *CacheRedisRepository) GetActiveCacheRegistration(ctx context.Context, logicalHostID string) (string, bool, error)

func (*CacheRedisRepository) GetCacheLogicalHost 

func (r *CacheRedisRepository) GetCacheLogicalHost(ctx context.Context, logicalHostID string) (cache.CoordinatorHost, bool, error)

func (*CacheRedisRepository) GetCacheRegistration 

func (r *CacheRedisRepository) GetCacheRegistration(ctx context.Context, logicalHostID, registrationID string) (cache.CoordinatorHost, bool, error)

func (*CacheRedisRepository) ListCacheLogicalHosts 

func (r *CacheRedisRepository) ListCacheLogicalHosts(ctx context.Context, poolName, locality string) ([]string, error)

func (*CacheRedisRepository) ListCacheRegistrations 

func (r *CacheRedisRepository) ListCacheRegistrations(ctx context.Context, logicalHostID string) ([]string, error)

func (*CacheRedisRepository) RemoveCacheLogicalHost 

func (r *CacheRedisRepository) RemoveCacheLogicalHost(ctx context.Context, poolName, locality, logicalHostID string) error

func (*CacheRedisRepository) RemoveCacheLogicalHostFromPool 

func (r *CacheRedisRepository) RemoveCacheLogicalHostFromPool(ctx context.Context, poolName, locality, logicalHostID string) error

func (*CacheRedisRepository) RemoveCacheRegistration 

func (r *CacheRedisRepository) RemoveCacheRegistration(ctx context.Context, logicalHostID, registrationID string) error

func (*CacheRedisRepository) SetActiveCacheRegistration 

func (r *CacheRedisRepository) SetActiveCacheRegistration(ctx context.Context, logicalHostID, registrationID string, ttl time.Duration) error

func (*CacheRedisRepository) SetCacheRegistration 

func (r *CacheRedisRepository) SetCacheRegistration(ctx context.Context, host cache.CoordinatorHost, ttl time.Duration) error

type ComputeRedisRepository 

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

func (*ComputeRedisRepository) DeleteAgentMachineState 

func (r *ComputeRedisRepository) DeleteAgentMachineState(ctx context.Context, workspaceID, poolName, machineID string) error

func (*ComputeRedisRepository) DeleteAgentWorkerSlotState 

func (r *ComputeRedisRepository) DeleteAgentWorkerSlotState(ctx context.Context, workspaceID, poolName, machineID, workerID string) error

func (*ComputeRedisRepository) DeleteFailoverDemand 

func (r *ComputeRedisRepository) DeleteFailoverDemand(ctx context.Context, gpu string) error

func (*ComputeRedisRepository) DeleteMachineSSHState 

func (r *ComputeRedisRepository) DeleteMachineSSHState(ctx context.Context, workspaceID, poolName, machineID string) error

func (*ComputeRedisRepository) DeletePoolState 

func (r *ComputeRedisRepository) DeletePoolState(ctx context.Context, workspaceID, name string) error

func (*ComputeRedisRepository) GetAgentMachineState 

func (r *ComputeRedisRepository) GetAgentMachineState(ctx context.Context, workspaceID, poolName, machineID string) (*compute.AgentTokenState, error)

func (*ComputeRedisRepository) GetAgentMachineStateForWorkspace 

func (r *ComputeRedisRepository) GetAgentMachineStateForWorkspace(ctx context.Context, workspaceID, machineID string) (*compute.AgentTokenState, error)

func (*ComputeRedisRepository) GetAgentTokenState 

func (r *ComputeRedisRepository) GetAgentTokenState(ctx context.Context, tokenHash string) (*compute.AgentTokenState, error)

func (*ComputeRedisRepository) GetJoinTokenState 

func (r *ComputeRedisRepository) GetJoinTokenState(ctx context.Context, tokenHash string) (*compute.JoinTokenState, error)

func (*ComputeRedisRepository) GetMachineSSHState 

func (r *ComputeRedisRepository) GetMachineSSHState(ctx context.Context, workspaceID, poolName, machineID string) (*compute.MachineSSHState, error)

func (*ComputeRedisRepository) GetPoolState 

func (r *ComputeRedisRepository) GetPoolState(ctx context.Context, workspaceID, name string) (*compute.PoolState, error)

func (*ComputeRedisRepository) ListAgentTokenStates 

func (r *ComputeRedisRepository) ListAgentTokenStates(ctx context.Context, workspaceID, poolName string) ([]*compute.AgentTokenState, error)

func (*ComputeRedisRepository) ListAgentWorkerSlotStates 

func (r *ComputeRedisRepository) ListAgentWorkerSlotStates(ctx context.Context, workspaceID, poolName, machineID string) ([]*compute.AgentWorkerSlotState, error)

func (*ComputeRedisRepository) ListAllPoolStates 

func (r *ComputeRedisRepository) ListAllPoolStates(ctx context.Context, limit int) ([]*compute.PoolState, error)

func (*ComputeRedisRepository) ListFailoverDemand 

func (r *ComputeRedisRepository) ListFailoverDemand(ctx context.Context) ([]*compute.FailoverDemand, error)

ListFailoverDemand returns the unexpired demand records and prunes index entries whose records have already expired.

func (*ComputeRedisRepository) ListPoolStates 

func (r *ComputeRedisRepository) ListPoolStates(ctx context.Context, workspaceID string, limit int) ([]*compute.PoolState, error)

func (*ComputeRedisRepository) OnDemandSpendCents 

func (r *ComputeRedisRepository) OnDemandSpendCents(ctx context.Context, window time.Duration) (float64, error)

OnDemandSpendCents sums the hourly buckets covering the trailing window.

func (*ComputeRedisRepository) PruneAgentMachineIndex 

func (r *ComputeRedisRepository) PruneAgentMachineIndex(ctx context.Context, workspaceID, poolName string) error

PruneAgentMachineIndex removes index entries whose machine state key no longer exists.

func (*ComputeRedisRepository) PushFailoverDemand 

func (r *ComputeRedisRepository) PushFailoverDemand(ctx context.Context, demand *compute.FailoverDemand, ttl time.Duration) error

func (*ComputeRedisRepository) RecordOnDemandSpend 

func (r *ComputeRedisRepository) RecordOnDemandSpend(ctx context.Context, at time.Time, cents float64) error

func (*ComputeRedisRepository) SaveAgentTokenState 

func (r *ComputeRedisRepository) SaveAgentTokenState(ctx context.Context, state *compute.AgentTokenState, ttl time.Duration) error

func (*ComputeRedisRepository) SaveAgentWorkerSlotState 

func (r *ComputeRedisRepository) SaveAgentWorkerSlotState(ctx context.Context, state *compute.AgentWorkerSlotState) error

func (*ComputeRedisRepository) SaveJoinTokenState 

func (r *ComputeRedisRepository) SaveJoinTokenState(ctx context.Context, state *compute.JoinTokenState, ttl time.Duration) error

func (*ComputeRedisRepository) SaveMachineSSHState 

func (r *ComputeRedisRepository) SaveMachineSSHState(ctx context.Context, state *compute.MachineSSHState) error

func (*ComputeRedisRepository) SavePoolState 

func (r *ComputeRedisRepository) SavePoolState(ctx context.Context, workspaceID string, state *compute.PoolState) error

func (*ComputeRedisRepository) WithMachineSSHStateLock 

func (r *ComputeRedisRepository) WithMachineSSHStateLock(ctx context.Context, workspaceID, poolName, machineID string, fn func(context.Context) error) error

func (*ComputeRedisRepository) WithPoolStateLock 

func (r *ComputeRedisRepository) WithPoolStateLock(ctx context.Context, workspaceID, name string, fn func(context.Context) error) error

type ComputeRepository 

type ComputeRepository interface {
	WithPoolStateLock(ctx context.Context, workspaceID, name string, fn func(context.Context) error) error
	SavePoolState(ctx context.Context, workspaceID string, state *compute.PoolState) error
	GetPoolState(ctx context.Context, workspaceID, name string) (*compute.PoolState, error)
	ListPoolStates(ctx context.Context, workspaceID string, limit int) ([]*compute.PoolState, error)
	ListAllPoolStates(ctx context.Context, limit int) ([]*compute.PoolState, error)
	DeletePoolState(ctx context.Context, workspaceID, name string) error
	SaveJoinTokenState(ctx context.Context, state *compute.JoinTokenState, ttl time.Duration) error
	GetJoinTokenState(ctx context.Context, tokenHash string) (*compute.JoinTokenState, error)
	SaveAgentTokenState(ctx context.Context, state *compute.AgentTokenState, ttl time.Duration) error
	GetAgentTokenState(ctx context.Context, tokenHash string) (*compute.AgentTokenState, error)
	GetAgentMachineState(ctx context.Context, workspaceID, poolName, machineID string) (*compute.AgentTokenState, error)
	GetAgentMachineStateForWorkspace(ctx context.Context, workspaceID, machineID string) (*compute.AgentTokenState, error)
	ListAgentTokenStates(ctx context.Context, workspaceID, poolName string) ([]*compute.AgentTokenState, error)
	DeleteAgentMachineState(ctx context.Context, workspaceID, poolName, machineID string) error
	PruneAgentMachineIndex(ctx context.Context, workspaceID, poolName string) error
	WithMachineSSHStateLock(ctx context.Context, workspaceID, poolName, machineID string, fn func(context.Context) error) error
	SaveMachineSSHState(ctx context.Context, state *compute.MachineSSHState) error
	GetMachineSSHState(ctx context.Context, workspaceID, poolName, machineID string) (*compute.MachineSSHState, error)
	DeleteMachineSSHState(ctx context.Context, workspaceID, poolName, machineID string) error
	SaveAgentWorkerSlotState(ctx context.Context, state *compute.AgentWorkerSlotState) error
	ListAgentWorkerSlotStates(ctx context.Context, workspaceID, poolName, machineID string) ([]*compute.AgentWorkerSlotState, error)
	DeleteAgentWorkerSlotState(ctx context.Context, workspaceID, poolName, machineID, workerID string) error
	PushFailoverDemand(ctx context.Context, demand *compute.FailoverDemand, ttl time.Duration) error
	ListFailoverDemand(ctx context.Context) ([]*compute.FailoverDemand, error)
	DeleteFailoverDemand(ctx context.Context, gpu string) error
	RecordOnDemandSpend(ctx context.Context, at time.Time, cents float64) error
	OnDemandSpendCents(ctx context.Context, window time.Duration) (float64, error)
}

func NewComputeRedisRepository 

func NewComputeRedisRepository(rdb *common.RedisClient) ComputeRepository

type ContainerRedisRepository 

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

func (*ContainerRedisRepository) AcknowledgeContainerExitCode 

func (cr *ContainerRedisRepository) AcknowledgeContainerExitCode(containerId string, exitCode int) error

AcknowledgeContainerExitCode releases retained VM proof only after the VM controller has saved a state that no longer owns this exact container. A crash before acknowledgement leaves harmless retained metadata; expiring unconsumed proof would instead make final disk sync impossible to confirm.

func (*ContainerRedisRepository) AcquireEndpointRequestToken 

func (c *ContainerRedisRepository) AcquireEndpointRequestToken(ctx context.Context, workspaceName, stubId, containerId string, maxTokens int, ttl time.Duration) (bool, error)

func (*ContainerRedisRepository) CheckContainerConcurrencyLimit 

func (c *ContainerRedisRepository) CheckContainerConcurrencyLimit(quota *types.ConcurrencyLimit, request *types.ContainerRequest) error

func (*ContainerRedisRepository) CreateContainerStateWithConcurrencyLimit 

func (c *ContainerRedisRepository) CreateContainerStateWithConcurrencyLimit(quota *types.ConcurrencyLimit, request *types.ContainerRequest) error

func (*ContainerRedisRepository) DeleteBackendRoutesByContainerID 

func (cr *ContainerRedisRepository) DeleteBackendRoutesByContainerID(ctx context.Context, containerID string) error

func (*ContainerRedisRepository) DeleteBackendRoutesByMachine 

func (cr *ContainerRedisRepository) DeleteBackendRoutesByMachine(ctx context.Context, workspaceID, poolName, machineID string) error

DeleteBackendRoutesByMachine removes all backend routes and index entries for a machine when it is released.

func (*ContainerRedisRepository) DeleteContainerState 

func (cr *ContainerRedisRepository) DeleteContainerState(containerId string) error

func (*ContainerRedisRepository) DeleteStubState 

func (cr *ContainerRedisRepository) DeleteStubState(stubId string) error

func (*ContainerRedisRepository) EndpointRequestHeartbeatExists 

func (c *ContainerRedisRepository) EndpointRequestHeartbeatExists(ctx context.Context, workspaceName, stubId, taskId, containerId string) (bool, error)

func (*ContainerRedisRepository) GetActiveContainersByStubId 

func (cr *ContainerRedisRepository) GetActiveContainersByStubId(stubId string) ([]types.ContainerState, error)

func (*ContainerRedisRepository) GetActiveContainersByWorkerId 

func (cr *ContainerRedisRepository) GetActiveContainersByWorkerId(workerId string) ([]types.ContainerState, error)

func (*ContainerRedisRepository) GetActiveContainersByWorkspaceId 

func (cr *ContainerRedisRepository) GetActiveContainersByWorkspaceId(workspaceId string) ([]types.ContainerState, error)

func (*ContainerRedisRepository) GetBackendRoute 

func (cr *ContainerRedisRepository) GetBackendRoute(ctx context.Context, routeID string) (*types.BackendRoute, error)

func (*ContainerRedisRepository) GetContainerAddress 

func (cr *ContainerRedisRepository) GetContainerAddress(containerId string) (string, error)

func (*ContainerRedisRepository) GetContainerAddressMap 

func (cr *ContainerRedisRepository) GetContainerAddressMap(containerId string) (map[int32]string, error)

func (*ContainerRedisRepository) GetContainerAddressMaps 

func (cr *ContainerRedisRepository) GetContainerAddressMaps(containerIds []string) (map[string]map[int32]string, error)

GetContainerAddressMaps is GetContainerAddressMap for many containers in one round trip. A container without an address map is absent from the result. When some cannot be read, the others are returned with the error of one of them, so that one bad address map does not hide the rest.

func (*ContainerRedisRepository) GetContainerExitCode 

func (cr *ContainerRedisRepository) GetContainerExitCode(containerId string) (int, error)

func (*ContainerRedisRepository) GetContainerRequestStatus 

func (cr *ContainerRedisRepository) GetContainerRequestStatus(containerId string) (types.ContainerRequestStatus, error)

func (*ContainerRedisRepository) GetContainerState 

func (cr *ContainerRedisRepository) GetContainerState(containerId string) (*types.ContainerState, error)

func (*ContainerRedisRepository) GetContainerStatuses 

func (cr *ContainerRedisRepository) GetContainerStatuses(containerIds []string) (map[string]types.ContainerStatus, error)

func (*ContainerRedisRepository) GetContainersByStubId 

func (cr *ContainerRedisRepository) GetContainersByStubId(stubId string) ([]types.ContainerState, []string, error)

GetContainersByStubId returns what GetActiveContainersByStubId and GetFailedContainersByStubId return, from one read of the stub's index and one round trip for every state and exit code, where the two together read the index twice and make four round trips. A stub's instance asks for both each time its autoscaler samples it, and for a stub of one sandbox the round trips are what the sample costs Redis.

func (*ContainerRedisRepository) GetEndpointRequestTokens 

func (c *ContainerRedisRepository) GetEndpointRequestTokens(ctx context.Context, workspaceName, stubId, containerId string, maxTokens int, ttl time.Duration) (int, error)

func (*ContainerRedisRepository) GetFailedContainersByStubId 

func (cr *ContainerRedisRepository) GetFailedContainersByStubId(stubId string) ([]string, error)

func (*ContainerRedisRepository) GetSandboxActivity 

func (cr *ContainerRedisRepository) GetSandboxActivity(workspaceId string, appIds []string, since time.Time) (map[string][]types.AppActivityBucket, error)

func (*ContainerRedisRepository) GetStubState 

func (cr *ContainerRedisRepository) GetStubState(stubId string) (string, error)

func (*ContainerRedisRepository) GetWorkerAddress 

func (cr *ContainerRedisRepository) GetWorkerAddress(ctx context.Context, containerId string) (string, error)

func (*ContainerRedisRepository) ListBackendRoutesByMachine 

func (cr *ContainerRedisRepository) ListBackendRoutesByMachine(ctx context.Context, workspaceID, poolName, machineID string) ([]types.BackendRoute, error)

func (*ContainerRedisRepository) ListBackendRoutesByMachineID 

func (cr *ContainerRedisRepository) ListBackendRoutesByMachineID(ctx context.Context, machineID string) ([]types.BackendRoute, error)

func (*ContainerRedisRepository) MarkPendingContainerStoppingIfUnassigned 

func (cr *ContainerRedisRepository) MarkPendingContainerStoppingIfUnassigned(containerId string, expirySeconds int64, exitCode int) (bool, error)

func (*ContainerRedisRepository) PodKeepWarmLockExists 

func (c *ContainerRedisRepository) PodKeepWarmLockExists(ctx context.Context, workspaceName, stubId, containerId string) (bool, error)

func (*ContainerRedisRepository) PodKeepWarmLocksExist 

func (c *ContainerRedisRepository) PodKeepWarmLocksExist(ctx context.Context, workspaceName, stubId string, containerIds []string) (map[string]bool, error)

PodKeepWarmLocksExist is PodKeepWarmLockExists for every container of a stub in one round trip. Containers without a lock are absent from the result.

func (*ContainerRedisRepository) RecordSandboxCreated 

func (cr *ContainerRedisRepository) RecordSandboxCreated(workspaceId, appId string, at time.Time) error

func (*ContainerRedisRepository) RefreshBuildContainerTTL 

func (cr *ContainerRedisRepository) RefreshBuildContainerTTL(containerId string, ttl time.Duration) (bool, error)

func (*ContainerRedisRepository) RefreshEndpointRequestTokenTTL 

func (c *ContainerRedisRepository) RefreshEndpointRequestTokenTTL(ctx context.Context, workspaceName, stubId, containerId string, ttl time.Duration) error

func (*ContainerRedisRepository) ReleaseEndpointRequestToken 

func (c *ContainerRedisRepository) ReleaseEndpointRequestToken(ctx context.Context, workspaceName, stubId, containerId, taskId string, maxTokens int, ttl time.Duration) error

func (*ContainerRedisRepository) RenewContainerLeases 

func (cr *ContainerRedisRepository) RenewContainerLeases(ctx context.Context, workerId, workspaceId string, renewals []types.ContainerLeaseRenewal) []types.ContainerLease

RenewContainerLeases is the routine status heartbeat for many containers of one worker. It costs one round trip to Redis whatever their number, where GetContainerState followed by UpdateContainerStatus takes a lock and several round trips for each container. workspaceId is empty for a worker that serves every workspace.

func (*ContainerRedisRepository) ReserveContainerConcurrencyForPending 

func (c *ContainerRedisRepository) ReserveContainerConcurrencyForPending(quota *types.ConcurrencyLimit, request *types.ContainerRequest) error

ReserveContainerConcurrencyForPending applies workspace quota accounting to an already-created PENDING container without changing its state or TTL. Initial admissions, including serverless, use CreateContainerStateWithConcurrencyLimit; this transition is for a quota-exempt request rerouted to accounted capacity.

func (*ContainerRedisRepository) SetBackendRoute 

func (cr *ContainerRedisRepository) SetBackendRoute(ctx context.Context, route types.BackendRoute) error

func (*ContainerRedisRepository) SetBackendRoutes 

func (cr *ContainerRedisRepository) SetBackendRoutes(ctx context.Context, routes []types.BackendRoute) error

func (*ContainerRedisRepository) SetBuildContainerTTL 

func (cr *ContainerRedisRepository) SetBuildContainerTTL(containerId string, ttl time.Duration) error

func (*ContainerRedisRepository) SetContainerAddress 

func (cr *ContainerRedisRepository) SetContainerAddress(containerId string, addr string) error

func (*ContainerRedisRepository) SetContainerAddressMap 

func (cr *ContainerRedisRepository) SetContainerAddressMap(containerId string, addressMap map[int32]string) error

func (*ContainerRedisRepository) SetContainerAddresses 

func (cr *ContainerRedisRepository) SetContainerAddresses(containerId string, addressMap map[int32]string, addr string) error

SetContainerAddresses is SetContainerAddressMap followed by SetContainerAddress, sent together.

func (*ContainerRedisRepository) SetContainerExitCode 

func (cr *ContainerRedisRepository) SetContainerExitCode(containerId string, exitCode int) error

func (*ContainerRedisRepository) SetContainerFailureCooldown 

func (cr *ContainerRedisRepository) SetContainerFailureCooldown(containerIds []string) error

func (*ContainerRedisRepository) SetContainerRequestStatus 

func (cr *ContainerRedisRepository) SetContainerRequestStatus(containerId string, status types.ContainerRequestStatus) error

func (*ContainerRedisRepository) SetContainerState 

func (cr *ContainerRedisRepository) SetContainerState(containerId string, state *types.ContainerState) error

func (*ContainerRedisRepository) SetEndpointRequestHeartbeat 

func (c *ContainerRedisRepository) SetEndpointRequestHeartbeat(ctx context.Context, workspaceName, stubId, taskId, containerId string, ttl time.Duration) error

func (*ContainerRedisRepository) SetPodKeepWarmLock 

func (c *ContainerRedisRepository) SetPodKeepWarmLock(ctx context.Context, workspaceName, stubId, containerId string, keepWarmSeconds int) error

func (*ContainerRedisRepository) SetStubState 

func (cr *ContainerRedisRepository) SetStubState(stubId, state string) error

func (*ContainerRedisRepository) SetWorkerAddress 

func (cr *ContainerRedisRepository) SetWorkerAddress(containerId string, addr string) error

func (*ContainerRedisRepository) UpdateContainerStatus 

func (cr *ContainerRedisRepository) UpdateContainerStatus(containerId string, requestedStatus types.ContainerStatus, expirySeconds int64) error

func (*ContainerRedisRepository) UpdateContainerStatusAndGetState 

func (cr *ContainerRedisRepository) UpdateContainerStatusAndGetState(containerId string, requestedStatus types.ContainerStatus, expirySeconds int64) (*types.ContainerState, error)

UpdateContainerStatusAndGetState is UpdateContainerStatus followed by GetContainerState, for a caller that acts on what the update left. A refused transition is not an error, so the state may not have the requested status.

type ContainerRepository 

type ContainerRepository interface {
	GetContainerState(string) (*types.ContainerState, error)
	SetContainerState(string, *types.ContainerState) error
	SetContainerExitCode(string, int) error
	SetContainerFailureCooldown([]string) error
	GetContainerExitCode(string) (int, error)
	AcknowledgeContainerExitCode(string, int) error
	SetContainerAddress(containerId string, addr string) error
	GetContainerAddress(containerId string) (string, error)
	SetBackendRoute(ctx context.Context, route types.BackendRoute) error
	SetBackendRoutes(ctx context.Context, routes []types.BackendRoute) error
	GetBackendRoute(ctx context.Context, routeID string) (*types.BackendRoute, error)
	ListBackendRoutesByMachine(ctx context.Context, workspaceID, poolName, machineID string) ([]types.BackendRoute, error)
	ListBackendRoutesByMachineID(ctx context.Context, machineID string) ([]types.BackendRoute, error)
	DeleteBackendRoutesByContainerID(ctx context.Context, containerID string) error
	DeleteBackendRoutesByMachine(ctx context.Context, workspaceID, poolName, machineID string) error
	UpdateContainerStatus(string, types.ContainerStatus, int64) error
	RenewContainerLeases(ctx context.Context, workerId, workspaceId string, renewals []types.ContainerLeaseRenewal) []types.ContainerLease
	MarkPendingContainerStoppingIfUnassigned(containerId string, expirySeconds int64, exitCode int) (bool, error)
	DeleteContainerState(containerId string) error
	SetContainerRequestStatus(containerId string, status types.ContainerRequestStatus) error
	GetContainerRequestStatus(containerId string) (types.ContainerRequestStatus, error)
	SetWorkerAddress(containerId string, addr string) error
	GetWorkerAddress(ctx context.Context, containerId string) (string, error)
	SetContainerAddressMap(containerId string, addressMap map[int32]string) error
	GetContainerAddressMap(containerId string) (map[int32]string, error)
	GetContainerAddressMaps(containerIds []string) (map[string]map[int32]string, error)
	CheckContainerConcurrencyLimit(quota *types.ConcurrencyLimit, request *types.ContainerRequest) error
	CreateContainerStateWithConcurrencyLimit(quota *types.ConcurrencyLimit, request *types.ContainerRequest) error
	ReserveContainerConcurrencyForPending(quota *types.ConcurrencyLimit, request *types.ContainerRequest) error
	GetActiveContainersByStubId(stubId string) ([]types.ContainerState, error)
	GetActiveContainersByWorkspaceId(workspaceId string) ([]types.ContainerState, error)
	RecordSandboxCreated(workspaceId, appId string, at time.Time) error
	GetSandboxActivity(workspaceId string, appIds []string, since time.Time) (map[string][]types.AppActivityBucket, error)
	GetActiveContainersByWorkerId(workerId string) ([]types.ContainerState, error)
	GetFailedContainersByStubId(stubId string) ([]string, error)
	GetContainersByStubId(stubId string) (active []types.ContainerState, failed []string, err error)
	GetStubState(stubId string) (string, error)
	SetStubState(stubId, state string) error
	DeleteStubState(stubId string) error
	SetBuildContainerTTL(containerId string, ttl time.Duration) error
	RefreshBuildContainerTTL(containerId string, ttl time.Duration) (bool, error)
	GetEndpointRequestTokens(ctx context.Context, workspaceName, stubId, containerId string, maxTokens int, ttl time.Duration) (int, error)
	AcquireEndpointRequestToken(ctx context.Context, workspaceName, stubId, containerId string, maxTokens int, ttl time.Duration) (bool, error)
	ReleaseEndpointRequestToken(ctx context.Context, workspaceName, stubId, containerId, taskId string, maxTokens int, ttl time.Duration) error
	RefreshEndpointRequestTokenTTL(ctx context.Context, workspaceName, stubId, containerId string, ttl time.Duration) error
	SetEndpointRequestHeartbeat(ctx context.Context, workspaceName, stubId, taskId, containerId string, ttl time.Duration) error
	EndpointRequestHeartbeatExists(ctx context.Context, workspaceName, stubId, taskId, containerId string) (bool, error)
	SetPodKeepWarmLock(ctx context.Context, workspaceName, stubId, containerId string, keepWarmSeconds int) error
	PodKeepWarmLockExists(ctx context.Context, workspaceName, stubId, containerId string) (bool, error)
	PodKeepWarmLocksExist(ctx context.Context, workspaceName, stubId string, containerIds []string) (map[string]bool, error)
}

func NewContainerRedisRepository 

func NewContainerRedisRepository(r *common.RedisClient) ContainerRepository

func NewContainerRedisRepositoryForTest 

func NewContainerRedisRepositoryForTest(rdb *common.RedisClient) ContainerRepository

type EventClientOption 

type EventClientOption func(*EventClientRepo)

EventClientOption adds sinks beyond the configured ones.

func WithWorkspaceWebhooks 

func WithWorkspaceWebhooks(source WebhookSource) EventClientOption

WithWorkspaceWebhooks fans events out to workspace webhooks. Gateway only.

func WithoutQueueWait 

func WithoutQueueWait() EventClientOption

WithoutQueueWait drops an event at once when the event store's queue is full, where the default is to hold the caller briefly for room. It is for callers that must not slow down when events are produced faster than they can be stored, such as the scheduling loop: held up, it placed containers only as fast as their timings could be written.

type EventClientRepo 

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

func (*EventClientRepo) GetContainerEvents 

func (r *EventClientRepo) GetContainerEvents(ctx context.Context, containerID string, query types.EventQuery) (*types.ContainerEventsResponse, error)

func (*EventClientRepo) GetEventHistory 

func (r *EventClientRepo) GetEventHistory(ctx context.Context, query types.EventQuery) (*types.EventHistoryResponse, error)

func (*EventClientRepo) GetLogs 

func (r *EventClientRepo) GetLogs(ctx context.Context, query types.LogQuery) (*types.LogsResponse, error)

func (*EventClientRepo) GetPoolMetricsTimeseries 

func (r *EventClientRepo) GetPoolMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.PoolMetricsTimeseriesResponse, error)

func (*EventClientRepo) GetStubMetricsTimeseries 

func (r *EventClientRepo) GetStubMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.MetricsTimeseriesResponse, error)

func (*EventClientRepo) GetWorkspaceMetricsTimeseries 

func (r *EventClientRepo) GetWorkspaceMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.MetricsTimeseriesResponse, error)

func (*EventClientRepo) PushCloneStubEvent 

func (r *EventClientRepo) PushCloneStubEvent(workspaceId string, stub *types.Stub, parentStub *types.Stub, actor types.EventActor)

func (*EventClientRepo) PushComputeEvent 

func (r *EventClientRepo) PushComputeEvent(eventType string, event types.EventComputeSchema)

func (*EventClientRepo) PushContainerEvent 

func (r *EventClientRepo) PushContainerEvent(event types.EventContainerEventSchema)

func (*EventClientRepo) PushContainerFunctionTaskEvent 

func (r *EventClientRepo) PushContainerFunctionTaskEvent(workspaceID string, task types.TaskInterface, eventID types.ContainerEventID, opts types.ContainerEventOptions)

func (*EventClientRepo) PushContainerFunctionTaskLifecycle 

func (r *EventClientRepo) PushContainerFunctionTaskLifecycle(workspaceID string, task types.TaskInterface, lifecycleID types.ContainerLifecycleID, start time.Time, end time.Time, success bool, opts types.ContainerLifecycleOptions)

func (*EventClientRepo) PushContainerLifecycleEvent 

func (r *EventClientRepo) PushContainerLifecycleEvent(lifecycle types.EventContainerLifecycleSchema)

func (*EventClientRepo) PushContainerLogDropped 

func (r *EventClientRepo) PushContainerLogDropped(workerID string, request *types.ContainerRequest, message types.EventMessage, taskID string)

func (*EventClientRepo) PushContainerLogEvent 

func (r *EventClientRepo) PushContainerLogEvent(entry types.EventContainerLogSchema)

func (*EventClientRepo) PushContainerLogEventQueued 

func (r *EventClientRepo) PushContainerLogEventQueued(entry types.EventContainerLogSchema) error

func (*EventClientRepo) PushContainerLogFirstByte 

func (r *EventClientRepo) PushContainerLogFirstByte(workerID string, request *types.ContainerRequest, taskID string)

func (*EventClientRepo) PushContainerLogFlushCompleted 

func (r *EventClientRepo) PushContainerLogFlushCompleted(workerID string, request *types.ContainerRequest)

func (*EventClientRepo) PushContainerLogLastByte 

func (r *EventClientRepo) PushContainerLogLastByte(workerID string, request *types.ContainerRequest)

func (*EventClientRepo) PushContainerRequestEvent 

func (r *EventClientRepo) PushContainerRequestEvent(workerID string, request *types.ContainerRequest, eventID types.ContainerEventID, opts types.ContainerEventOptions)

func (*EventClientRepo) PushContainerRequestLifecycle 

func (r *EventClientRepo) PushContainerRequestLifecycle(workerID string, request *types.ContainerRequest, lifecycleID types.ContainerLifecycleID, startedAt time.Time, duration time.Duration, success bool, opts types.ContainerLifecycleOptions)

func (*EventClientRepo) PushContainerRequestLogLine 

func (r *EventClientRepo) PushContainerRequestLogLine(workerID string, request *types.ContainerRequest, taskID string, stream string, line string)

func (*EventClientRepo) PushContainerResourceMetricsEvent 

func (r *EventClientRepo) PushContainerResourceMetricsEvent(workerID string, request *types.ContainerRequest, metrics types.EventContainerMetricsData)

func (*EventClientRepo) PushContainerRunnerEvent 

func (r *EventClientRepo) PushContainerRunnerEvent(workerID string, request *types.ContainerRequest, event *types.ContainerRunnerEvent)

func (*EventClientRepo) PushContainerRunningToStartTask 

func (r *EventClientRepo) PushContainerRunningToStartTask(task *types.TaskWithRelated, runningAt time.Time, startedAt time.Time, status types.ContainerStatus)

func (*EventClientRepo) PushContainerTaskEvent 

func (r *EventClientRepo) PushContainerTaskEvent(task *types.TaskWithRelated, eventID types.ContainerEventID, opts types.ContainerEventOptions)

func (*EventClientRepo) PushContainerTaskLifecycle 

func (r *EventClientRepo) PushContainerTaskLifecycle(task *types.TaskWithRelated, lifecycleID types.ContainerLifecycleID, start time.Time, end time.Time, success bool, opts types.ContainerLifecycleOptions)

func (*EventClientRepo) PushContainerTaskLifecycleSince 

func (r *EventClientRepo) PushContainerTaskLifecycleSince(ctx context.Context, rdb *common.RedisClient, task *types.TaskWithRelated, lifecycleID types.ContainerLifecycleID, sincePhase string, end time.Time, success bool, opts types.ContainerLifecycleOptions)

func (*EventClientRepo) PushDeployStubEvent 

func (r *EventClientRepo) PushDeployStubEvent(workspaceId string, stub *types.Stub, actor types.EventActor)

func (*EventClientRepo) PushEndpointEvent 

func (r *EventClientRepo) PushEndpointEvent(eventType string, event types.EventEndpointSchema)

func (*EventClientRepo) PushEndpointRequestStatsEvent 

func (r *EventClientRepo) PushEndpointRequestStatsEvent(schema types.EventEndpointRequestStatsSchema)

func (*EventClientRepo) PushEndpointRouteEvent 

func (r *EventClientRepo) PushEndpointRouteEvent(event types.EventEndpointRouteSchema)

func (*EventClientRepo) PushFunctionGetArgs 

func (r *EventClientRepo) PushFunctionGetArgs(ctx context.Context, rdb *common.RedisClient, task *types.TaskWithRelated, at time.Time, byteCount int)

func (*EventClientRepo) PushFunctionResultDelivery 

func (r *EventClientRepo) PushFunctionResultDelivery(workspaceID string, task types.TaskInterface, startedAt time.Time, exitCode int32, byteCount int)

func (*EventClientRepo) PushFunctionResultLoaded 

func (r *EventClientRepo) PushFunctionResultLoaded(workspaceID string, task types.TaskInterface, exitCode int32, byteCount int)

func (*EventClientRepo) PushFunctionResultSent 

func (r *EventClientRepo) PushFunctionResultSent(workspaceID string, task types.TaskInterface, exitCode int32, byteCount int)

func (*EventClientRepo) PushFunctionSetResult 

func (r *EventClientRepo) PushFunctionSetResult(ctx context.Context, rdb *common.RedisClient, task *types.TaskWithRelated, at time.Time, byteCount int)

func (*EventClientRepo) PushFunctionStreamCancelApplied 

func (r *EventClientRepo) PushFunctionStreamCancelApplied(workspaceID string, task types.TaskInterface)

func (*EventClientRepo) PushFunctionStreamCancelRequested 

func (r *EventClientRepo) PushFunctionStreamCancelRequested(workspaceID string, task types.TaskInterface)

func (*EventClientRepo) PushGatewayEndpointCalledEvent 

func (r *EventClientRepo) PushGatewayEndpointCalledEvent(method, path, workspaceID string, statusCode int, userAgent, remoteIP, requestID, contentType, accept, errorMessage string)

func (*EventClientRepo) PushPlatformCacheEvent 

func (r *EventClientRepo) PushPlatformCacheEvent(schema types.EventPlatformCacheSchema)

func (*EventClientRepo) PushPlatformLogEvent 

func (r *EventClientRepo) PushPlatformLogEvent(entry types.EventPlatformLogSchema)

func (*EventClientRepo) PushRunStubEvent 

func (r *EventClientRepo) PushRunStubEvent(workspaceId string, stub *types.Stub, actor types.EventActor)

func (*EventClientRepo) PushServeStubEvent 

func (r *EventClientRepo) PushServeStubEvent(workspaceId string, stub *types.Stub, actor types.EventActor)

func (*EventClientRepo) PushStubCacheRequiredContent 

func (r *EventClientRepo) PushStubCacheRequiredContent(schema types.EventStubCacheRequiredContentSchema) error

func (*EventClientRepo) PushStubStateUnhealthy 

func (r *EventClientRepo) PushStubStateUnhealthy(workspaceId string, stubId string, currentState string, previousState string, reason string, failedContainers []string)

func (*EventClientRepo) PushTaskCancelApplied 

func (r *EventClientRepo) PushTaskCancelApplied(task *types.TaskWithRelated, source types.EventSource, message types.EventMessage)

func (*EventClientRepo) PushTaskCancelRequested 

func (r *EventClientRepo) PushTaskCancelRequested(task *types.TaskWithRelated, source types.EventSource, message types.EventMessage)

func (*EventClientRepo) PushTaskCreatedEvent 

func (r *EventClientRepo) PushTaskCreatedEvent(task *types.TaskWithRelated)

func (*EventClientRepo) PushTaskEndEvents 

func (r *EventClientRepo) PushTaskEndEvents(ctx context.Context, rdb *common.RedisClient, task *types.TaskWithRelated, endedAt time.Time)

func (*EventClientRepo) PushTaskEndPersisted 

func (r *EventClientRepo) PushTaskEndPersisted(task *types.TaskWithRelated)

func (*EventClientRepo) PushTaskStartEvents 

func (r *EventClientRepo) PushTaskStartEvents(ctx context.Context, rdb *common.RedisClient, task *types.TaskWithRelated, containerID string, startedAt time.Time)

func (*EventClientRepo) PushTaskUpdatedEvent 

func (r *EventClientRepo) PushTaskUpdatedEvent(task *types.TaskWithRelated)

func (*EventClientRepo) PushWorkerDeletedEvent 

func (r *EventClientRepo) PushWorkerDeletedEvent(workerID, machineID, poolName string, reason types.DeletedWorkerReason)

func (*EventClientRepo) PushWorkerStartedEvent 

func (r *EventClientRepo) PushWorkerStartedEvent(workerID string)

func (*EventClientRepo) PushWorkerStoppedEvent 

func (r *EventClientRepo) PushWorkerStoppedEvent(workerID string)

func (*EventClientRepo) ReadStubCacheRequiredContent 

func (r *EventClientRepo) ReadStubCacheRequiredContent(ctx context.Context, workspaceID, stubID string) ([]types.CacheRequiredContentItem, error)

func (*EventClientRepo) StreamAppNamespaceEvents 

func (r *EventClientRepo) StreamAppNamespaceEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

func (*EventClientRepo) StreamContainerEvents 

func (r *EventClientRepo) StreamContainerEvents(ctx context.Context, containerID string, query types.EventQuery) (EventStream, error)

func (*EventClientRepo) StreamLogs 

func (r *EventClientRepo) StreamLogs(ctx context.Context, query types.LogQuery) (EventStream, error)

func (*EventClientRepo) StreamStubEvents 

func (r *EventClientRepo) StreamStubEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

func (*EventClientRepo) StreamTaskEvents 

func (r *EventClientRepo) StreamTaskEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

func (*EventClientRepo) StreamWorkspaceEvents 

func (r *EventClientRepo) StreamWorkspaceEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

type EventRepository 

type EventRepository interface {
	GetContainerEvents(ctx context.Context, containerID string, query types.EventQuery) (*types.ContainerEventsResponse, error)
	GetEventHistory(ctx context.Context, query types.EventQuery) (*types.EventHistoryResponse, error)
	GetLogs(ctx context.Context, query types.LogQuery) (*types.LogsResponse, error)
	GetStubMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.MetricsTimeseriesResponse, error)
	GetWorkspaceMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.MetricsTimeseriesResponse, error)
	GetPoolMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.PoolMetricsTimeseriesResponse, error)
	StreamContainerEvents(ctx context.Context, containerID string, query types.EventQuery) (EventStream, error)
	StreamStubEvents(ctx context.Context, query types.EventQuery) (EventStream, error)
	StreamTaskEvents(ctx context.Context, query types.EventQuery) (EventStream, error)
	StreamWorkspaceEvents(ctx context.Context, query types.EventQuery) (EventStream, error)
	StreamAppNamespaceEvents(ctx context.Context, query types.EventQuery) (EventStream, error)
	StreamLogs(ctx context.Context, query types.LogQuery) (EventStream, error)
	PushContainerResourceMetricsEvent(workerID string, request *types.ContainerRequest, metrics types.EventContainerMetricsData)
	PushContainerLifecycleEvent(lifecycle types.EventContainerLifecycleSchema)
	PushContainerEvent(event types.EventContainerEventSchema)
	PushContainerLogEvent(entry types.EventContainerLogSchema)
	PushContainerLogEventQueued(entry types.EventContainerLogSchema) error
	PushPlatformLogEvent(entry types.EventPlatformLogSchema)
	PushContainerRequestEvent(workerID string, request *types.ContainerRequest, eventID types.ContainerEventID, opts types.ContainerEventOptions)
	PushContainerRequestLifecycle(workerID string, request *types.ContainerRequest, lifecycleID types.ContainerLifecycleID, startedAt time.Time, duration time.Duration, success bool, opts types.ContainerLifecycleOptions)
	PushContainerTaskEvent(task *types.TaskWithRelated, eventID types.ContainerEventID, opts types.ContainerEventOptions)
	PushContainerFunctionTaskEvent(workspaceID string, task types.TaskInterface, eventID types.ContainerEventID, opts types.ContainerEventOptions)
	PushContainerTaskLifecycle(task *types.TaskWithRelated, lifecycleID types.ContainerLifecycleID, start time.Time, end time.Time, success bool, opts types.ContainerLifecycleOptions)
	PushContainerFunctionTaskLifecycle(workspaceID string, task types.TaskInterface, lifecycleID types.ContainerLifecycleID, start time.Time, end time.Time, success bool, opts types.ContainerLifecycleOptions)
	PushContainerTaskLifecycleSince(ctx context.Context, rdb *pkgcommon.RedisClient, task *types.TaskWithRelated, lifecycleID types.ContainerLifecycleID, sincePhase string, end time.Time, success bool, opts types.ContainerLifecycleOptions)
	PushContainerRequestLogLine(workerID string, request *types.ContainerRequest, taskID string, stream string, line string)
	PushContainerRunnerEvent(workerID string, request *types.ContainerRequest, event *types.ContainerRunnerEvent)
	PushFunctionResultLoaded(workspaceID string, task types.TaskInterface, exitCode int32, byteCount int)
	PushFunctionResultSent(workspaceID string, task types.TaskInterface, exitCode int32, byteCount int)
	PushFunctionResultDelivery(workspaceID string, task types.TaskInterface, startedAt time.Time, exitCode int32, byteCount int)
	PushFunctionStreamCancelRequested(workspaceID string, task types.TaskInterface)
	PushFunctionStreamCancelApplied(workspaceID string, task types.TaskInterface)
	PushFunctionGetArgs(ctx context.Context, rdb *pkgcommon.RedisClient, task *types.TaskWithRelated, at time.Time, byteCount int)
	PushFunctionSetResult(ctx context.Context, rdb *pkgcommon.RedisClient, task *types.TaskWithRelated, at time.Time, byteCount int)
	PushTaskStartEvents(ctx context.Context, rdb *pkgcommon.RedisClient, task *types.TaskWithRelated, containerID string, startedAt time.Time)
	PushTaskEndEvents(ctx context.Context, rdb *pkgcommon.RedisClient, task *types.TaskWithRelated, endedAt time.Time)
	PushTaskEndPersisted(task *types.TaskWithRelated)
	PushContainerRunningToStartTask(task *types.TaskWithRelated, runningAt time.Time, startedAt time.Time, status types.ContainerStatus)
	PushTaskCancelRequested(task *types.TaskWithRelated, source types.EventSource, message types.EventMessage)
	PushTaskCancelApplied(task *types.TaskWithRelated, source types.EventSource, message types.EventMessage)
	PushContainerLogFlushCompleted(workerID string, request *types.ContainerRequest)
	PushContainerLogDropped(workerID string, request *types.ContainerRequest, message types.EventMessage, taskID string)
	PushContainerLogFirstByte(workerID string, request *types.ContainerRequest, taskID string)
	PushContainerLogLastByte(workerID string, request *types.ContainerRequest)
	PushWorkerStartedEvent(workerID string)
	PushWorkerStoppedEvent(workerID string)
	PushWorkerDeletedEvent(workerID, machineID, poolName string, reason types.DeletedWorkerReason)
	PushComputeEvent(eventType string, event types.EventComputeSchema)
	PushEndpointEvent(eventType string, event types.EventEndpointSchema)
	PushEndpointRouteEvent(event types.EventEndpointRouteSchema)
	PushDeployStubEvent(workspaceId string, stub *types.Stub, actor types.EventActor)
	PushServeStubEvent(workspaceId string, stub *types.Stub, actor types.EventActor)
	PushRunStubEvent(workspaceId string, stub *types.Stub, actor types.EventActor)
	PushCloneStubEvent(workspaceId string, stub *types.Stub, parentStub *types.Stub, actor types.EventActor)
	PushTaskUpdatedEvent(task *types.TaskWithRelated)
	PushTaskCreatedEvent(task *types.TaskWithRelated)
	PushStubStateUnhealthy(workspaceId string, stubId string, currentState, previousState string, reason string, failedContainers []string)
	PushGatewayEndpointCalledEvent(method, path, workspaceID string, statusCode int, userAgent, remoteIP, requestID, contentType, accept, errorMessage string)
	PushEndpointRequestStatsEvent(schema types.EventEndpointRequestStatsSchema)
	PushStubCacheRequiredContent(schema types.EventStubCacheRequiredContentSchema) error
	PushPlatformCacheEvent(schema types.EventPlatformCacheSchema)
	ReadStubCacheRequiredContent(ctx context.Context, workspaceID, stubID string) ([]types.CacheRequiredContentItem, error)
}

func NewEventClientRepo 

func NewEventClientRepo(config types.AppConfig, opts ...EventClientOption) EventRepository

func NewWorkerEventClientRepo 

func NewWorkerEventClientRepo(config types.AppConfig, client pb.WorkerRepositoryServiceClient, workerID string) EventRepository

type EventStream 

type EventStream interface {
	Next() bool
	Record() types.ContainerEventRecord
	Err() error
	Close() error
}

type GooseLogger 

type GooseLogger struct {
	zerolog.Logger
}

func (*GooseLogger) Fatalf 

func (l *GooseLogger) Fatalf(format string, v ...any)

type ManagedEndpointRedisRepository 

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

ManagedEndpointRedisRepository implements ManagedEndpointRepository on Redis. Keys (under managed_endpoint:): 

app:<id>, replica:<rid>                           JSON, indexed by the sets apps, replicas, replicas:<id>
replica_container:<cid> -> rid; replica_lock:<rid>, drain:<rid>, backoff:<id>:<gpu>
fleet, gitops                                     JSON
generation:<charge_id>                            JSON charge; accounting:pending ZSET of ids not yet metered
config_events:<rid>                               pub/sub, replica config revision numbers
metrics:<id>:<gpu>:<replica[@revision]>:<minute>  HASH counters

func (*ManagedEndpointRedisRepository) CompleteAccounting 

func (r *ManagedEndpointRedisRepository) CompleteAccounting(ctx context.Context, id string, ttl time.Duration) error

CompleteAccounting takes a charge out of the pending index and lets its journal entry expire.

func (*ManagedEndpointRedisRepository) DeleteReplica 

func (r *ManagedEndpointRedisRepository) DeleteReplica(ctx context.Context, replicaID string) error

func (*ManagedEndpointRedisRepository) DrainRequested 

func (r *ManagedEndpointRedisRepository) DrainRequested(ctx context.Context, replicaID string) (bool, uint32, error)

func (*ManagedEndpointRedisRepository) GetCharge 

func (r *ManagedEndpointRedisRepository) GetCharge(ctx context.Context, id string) (*types.Charge, error)

GetCharge returns nil for an id with no charge, including a record written before charges existed.

func (*ManagedEndpointRedisRepository) GetEndpoint 

func (r *ManagedEndpointRedisRepository) GetEndpoint(ctx context.Context, id string) (*types.ManagedEndpoint, error)

func (*ManagedEndpointRedisRepository) GetFleet 

func (r *ManagedEndpointRedisRepository) GetFleet(ctx context.Context) (*types.Fleet, error)

GetFleet returns the stored fleet, or an empty one when none was applied yet.

func (*ManagedEndpointRedisRepository) GetGitOpsState 

func (r *ManagedEndpointRedisRepository) GetGitOpsState(ctx context.Context) (*types.GitOpsState, error)

func (*ManagedEndpointRedisRepository) GetReplica 

func (r *ManagedEndpointRedisRepository) GetReplica(ctx context.Context, replicaID string) (*types.EndpointReplica, error)

func (*ManagedEndpointRedisRepository) GetReplicaByContainer 

func (r *ManagedEndpointRedisRepository) GetReplicaByContainer(ctx context.Context, containerID string) (*types.EndpointReplica, error)

func (*ManagedEndpointRedisRepository) GetRouteMetrics 

func (r *ManagedEndpointRedisRepository) GetRouteMetrics(ctx context.Context, endpointID, gpu, replicaID string, configRevision uint64, window time.Duration) (*types.RouteMetrics, error)

GetRouteMetrics sums the window for the app, one GPU type, one replica or one replica under one config revision.

func (*ManagedEndpointRedisRepository) InScheduleBackoff 

func (r *ManagedEndpointRedisRepository) InScheduleBackoff(ctx context.Context, endpointID, gpu string) (bool, error)

func (*ManagedEndpointRedisRepository) ListAllReplicas 

func (r *ManagedEndpointRedisRepository) ListAllReplicas(ctx context.Context) ([]*types.EndpointReplica, error)

func (*ManagedEndpointRedisRepository) ListEndpoints 

func (r *ManagedEndpointRedisRepository) ListEndpoints(ctx context.Context) ([]*types.ManagedEndpoint, error)

func (*ManagedEndpointRedisRepository) ListPendingCharges 

func (r *ManagedEndpointRedisRepository) ListPendingCharges(ctx context.Context, now time.Time, limit int64) ([]*types.Charge, error)

ListPendingCharges returns the charges due for accounting by now, oldest first. An index entry with no charge behind it is dropped with a warning rather than blocking every charge behind it.

func (*ManagedEndpointRedisRepository) ListReplicas 

func (r *ManagedEndpointRedisRepository) ListReplicas(ctx context.Context, endpointID string) ([]*types.EndpointReplica, error)

func (*ManagedEndpointRedisRepository) NotifyReplicaConfig 

func (r *ManagedEndpointRedisRepository) NotifyReplicaConfig(ctx context.Context, replicaID string, revision uint64) error

func (*ManagedEndpointRedisRepository) RecordRouteSample 

func (r *ManagedEndpointRedisRepository) RecordRouteSample(ctx context.Context, c *types.Charge) error

RecordRouteSample folds one finished charge into the minute buckets for the app, its GPU type, the replica and the replica's config revision.

func (*ManagedEndpointRedisRepository) RequestDrain 

func (r *ManagedEndpointRedisRepository) RequestDrain(ctx context.Context, replicaID string, drainSeconds uint32) error

func (*ManagedEndpointRedisRepository) SaveCharge 

func (r *ManagedEndpointRedisRepository) SaveCharge(ctx context.Context, c *types.Charge) (bool, error)

SaveCharge journals a charge and reports whether it was written. A charge is final once journaled, so a duplicate completion writes nothing. Written charges stay in the pending index until CompleteAccounting.

func (*ManagedEndpointRedisRepository) SaveEndpoint 

func (r *ManagedEndpointRedisRepository) SaveEndpoint(ctx context.Context, app *types.ManagedEndpoint) error

func (*ManagedEndpointRedisRepository) SaveFleet 

func (r *ManagedEndpointRedisRepository) SaveFleet(ctx context.Context, fleet *types.Fleet) error

func (*ManagedEndpointRedisRepository) SaveGitOpsState 

func (r *ManagedEndpointRedisRepository) SaveGitOpsState(ctx context.Context, state *types.GitOpsState) error

func (*ManagedEndpointRedisRepository) SaveReplica 

func (r *ManagedEndpointRedisRepository) SaveReplica(ctx context.Context, replica *types.EndpointReplica) error

func (*ManagedEndpointRedisRepository) SetReplicaProtection 

func (r *ManagedEndpointRedisRepository) SetReplicaProtection(ctx context.Context, replicaID string, protected bool) (*types.EndpointReplica, error)

SetReplicaProtection changes an assigned replica without replacing its process. Scheduling already takes this worker lease and checks each victim's live flag; protection adds no work to the ordinary serverless scheduling path.

func (*ManagedEndpointRedisRepository) SetScheduleBackoff 

func (r *ManagedEndpointRedisRepository) SetScheduleBackoff(ctx context.Context, endpointID, gpu string, ttl time.Duration) error

func (*ManagedEndpointRedisRepository) SubscribeReplicaConfig 

func (r *ManagedEndpointRedisRepository) SubscribeReplicaConfig(ctx context.Context, replicaID string) (<-chan uint64, error)

func (*ManagedEndpointRedisRepository) WithReplicaLock 

func (r *ManagedEndpointRedisRepository) WithReplicaLock(ctx context.Context, replicaID string, fn func(context.Context) error) error

type ManagedEndpointRepository 

type ManagedEndpointRepository interface {
	// Registry: the deployed endpoints and the fleet (placement) from the repo
	SaveEndpoint(ctx context.Context, endpoint *types.ManagedEndpoint) error
	GetEndpoint(ctx context.Context, endpointID string) (*types.ManagedEndpoint, error)
	ListEndpoints(ctx context.Context) ([]*types.ManagedEndpoint, error)
	SaveFleet(ctx context.Context, fleet *types.Fleet) error
	GetFleet(ctx context.Context) (*types.Fleet, error)

	// Replicas
	SaveReplica(ctx context.Context, replica *types.EndpointReplica) error
	GetReplica(ctx context.Context, replicaID string) (*types.EndpointReplica, error)
	GetReplicaByContainer(ctx context.Context, containerID string) (*types.EndpointReplica, error)
	ListReplicas(ctx context.Context, endpointID string) ([]*types.EndpointReplica, error)
	ListAllReplicas(ctx context.Context) ([]*types.EndpointReplica, error)
	DeleteReplica(ctx context.Context, replicaID string) error
	WithReplicaLock(ctx context.Context, replicaID string, fn func(context.Context) error) error
	SetReplicaProtection(ctx context.Context, replicaID string, protected bool) (*types.EndpointReplica, error)
	RequestDrain(ctx context.Context, replicaID string, drainSeconds uint32) error
	DrainRequested(ctx context.Context, replicaID string) (bool, uint32, error)
	SetScheduleBackoff(ctx context.Context, endpointID, gpu string, ttl time.Duration) error
	InScheduleBackoff(ctx context.Context, endpointID, gpu string) (bool, error)
	// Live config: the replica record holds the config; these wake WatchConfig streams.
	NotifyReplicaConfig(ctx context.Context, replicaID string, revision uint64) error
	SubscribeReplicaConfig(ctx context.Context, replicaID string) (<-chan uint64, error)

	// GitOps
	SaveGitOpsState(ctx context.Context, state *types.GitOpsState) error
	GetGitOpsState(ctx context.Context) (*types.GitOpsState, error)

	// Route metrics (minute buckets, bounded retention), derived from settled charges
	RecordRouteSample(ctx context.Context, charge *types.Charge) error
	GetRouteMetrics(ctx context.Context, endpointID, gpu, replicaID string, configRevision uint64, window time.Duration) (*types.RouteMetrics, error)

	// Charge journal: the durable accounting outbox
	SaveCharge(ctx context.Context, charge *types.Charge) (written bool, err error)
	GetCharge(ctx context.Context, id string) (*types.Charge, error)
	ListPendingCharges(ctx context.Context, now time.Time, limit int64) ([]*types.Charge, error)
	CompleteAccounting(ctx context.Context, id string, ttl time.Duration) error
}

ManagedEndpointRepository stores the fast-changing state of the managed endpoints platform: the registry of deployed endpoint/service versions, live replicas, harness config revisions and acks, tuning experiments, rollout and GitOps state, and rolled-up route metrics. The durable stub itself lives in Postgres; everything here can be rebuilt from stubs + heartbeats.

func NewManagedEndpointRedisRepository 

func NewManagedEndpointRedisRepository(rdb *common.RedisClient) ManagedEndpointRepository

type ManagedPoolRedisRepository 

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

func (*ManagedPoolRedisRepository) DeleteManagedPoolState 

func (r *ManagedPoolRedisRepository) DeleteManagedPoolState(ctx context.Context, workspaceID, name string) error

func (*ManagedPoolRedisRepository) GetManagedPoolState 

func (r *ManagedPoolRedisRepository) GetManagedPoolState(ctx context.Context, workspaceID, name string) (*compute.PoolState, error)

func (*ManagedPoolRedisRepository) ListManagedPoolStates 

func (r *ManagedPoolRedisRepository) ListManagedPoolStates(ctx context.Context, workspaceID string, limit int) ([]*compute.PoolState, error)

func (*ManagedPoolRedisRepository) SaveManagedPoolState 

func (r *ManagedPoolRedisRepository) SaveManagedPoolState(ctx context.Context, workspaceID string, state *compute.PoolState) error

func (*ManagedPoolRedisRepository) WithManagedPoolStateLock 

func (r *ManagedPoolRedisRepository) WithManagedPoolStateLock(ctx context.Context, workspaceID, name string, fn func(context.Context) error) error

type ManagedPoolRepository 

type ManagedPoolRepository interface {
	WithManagedPoolStateLock(ctx context.Context, workspaceID, name string, fn func(context.Context) error) error
	SaveManagedPoolState(ctx context.Context, workspaceID string, state *compute.PoolState) error
	GetManagedPoolState(ctx context.Context, workspaceID, name string) (*compute.PoolState, error)
	ListManagedPoolStates(ctx context.Context, workspaceID string, limit int) ([]*compute.PoolState, error)
	DeleteManagedPoolState(ctx context.Context, workspaceID, name string) error
}

ManagedPoolRepository keeps managed pool definitions separate from tenant/private compute state. That boundary lets mixed gateway versions roll safely: replicas that predate managed pools cannot enumerate or rewrite these records through ComputeRepository.

func NewManagedPoolRedisRepository 

func NewManagedPoolRedisRepository(rdb *common.RedisClient) ManagedPoolRepository

type OrphanSweepStats 

type OrphanSweepStats struct {
	// Keys the sweep walked over.
	Scanned int
	// Keys and index members removed, by rule.
	Removed map[string]int
	// Orphans that were not old enough yet and are remembered for the next
	// sweep.
	Waiting int
}

OrphanSweepStats is what one sweep did.

type OrphanSweeper 

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

OrphanSweeper removes Redis keys left behind by containers and workers that no longer exist.

func NewOrphanSweeper 

func NewOrphanSweeper(rdb *common.RedisClient) *OrphanSweeper

func (*OrphanSweeper) Run 

func (s *OrphanSweeper) Run(ctx context.Context)

Run sweeps until ctx is cancelled: first a few minutes after the gateway starts, then once per interval.

func (*OrphanSweeper) Sweep 

func (s *OrphanSweeper) Sweep(ctx context.Context) (OrphanSweepStats, error)

Sweep walks the whole keyspace once, a batch at a time, and removes the orphans it finds. The caller holds the sweep lock.

func (*OrphanSweeper) SweepIfDue 

func (s *OrphanSweeper) SweepIfDue(ctx context.Context) (bool, error)

SweepIfDue runs one sweep unless another gateway replica is sweeping right now or one swept within the cooldown. It reports whether this call swept.

type PostgresBackendRepository 

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

func NewBackendPostgresRepository 

func NewBackendPostgresRepository(config types.PostgresConfig, eventRepo EventRepository) (*PostgresBackendRepository, error)

func (*PostgresBackendRepository) AggregateTaskActivityByApp 

func (c *PostgresBackendRepository) AggregateTaskActivityByApp(ctx context.Context, workspaceID uint, appExternalIDs []string, since time.Time) (map[string][]types.AppActivityBucket, error)

AggregateTaskActivityByApp buckets tasks created since `since` by app and hour, for a set of apps, in one query. Backs the dashboard's per-card activity strips, replacing one aggregate request per card.

func (*PostgresBackendRepository) AggregateTasksByTimeWindow 

func (c *PostgresBackendRepository) AggregateTasksByTimeWindow(ctx context.Context, filters types.TaskFilter) ([]types.TaskCountByTime, error)

func (*PostgresBackendRepository) AssignVMPublicIPv6 

func (r *PostgresBackendRepository) AssignVMPublicIPv6(ctx context.Context, id, address string) (string, error)

Separate ownership from launch JSON: a stale lifecycle save cannot release an address, and retrying an ambiguous allocation returns the same reservation.

func (*PostgresBackendRepository) AuthorizeToken 

func (r *PostgresBackendRepository) AuthorizeToken(ctx context.Context, tokenKey string) (*types.Token, *types.Workspace, error)

func (*PostgresBackendRepository) BindWorkerToken 

func (r *PostgresBackendRepository) BindWorkerToken(ctx context.Context, tokenKey, workerId string) (bool, error)

BindWorkerToken binds an unbound worker token to the worker it was issued for. It reports whether the token is now bound to workerId.

func (*PostgresBackendRepository) ClaimVMIdleStop 

func (r *PostgresBackendRepository) ClaimVMIdleStop(ctx context.Context, id string, cutoff time.Time) (bool, error)

func (*PostgresBackendRepository) CountActiveDeploymentsByApp 

func (c *PostgresBackendRepository) CountActiveDeploymentsByApp(ctx context.Context, workspaceID uint, appExternalIDs []string) (map[string]int, error)

CountActiveDeploymentsByApp returns, per app, how many of its deployments are currently active. Unlike ListLatestDeploymentsByAppIDs this considers every deployment in the app, so an app whose newest deployment was stopped while an older function is still deployed reports as idle rather than stopped. A nil appExternalIDs covers every app in the workspace, which is what the dashboard needs to compute workspace-wide state counts; apps with no active deployment are simply absent from the result.

func (*PostgresBackendRepository) CountApps 

func (r *PostgresBackendRepository) CountApps(ctx context.Context, workspaceId uint) (int, error)

CountApps returns the number of live (non-deleted) apps in a workspace.

func (*PostgresBackendRepository) CreateCheckpoint 

func (r *PostgresBackendRepository) CreateCheckpoint(ctx context.Context, checkpoint *types.Checkpoint, compatibilityKey string) (*types.Checkpoint, error)

func (*PostgresBackendRepository) CreateConcurrencyLimit 

func (r *PostgresBackendRepository) CreateConcurrencyLimit(ctx context.Context, workspaceId uint, gpuLimit uint32, cpuMillicoreLimit uint32) (*types.ConcurrencyLimit, error)

func (*PostgresBackendRepository) CreateDeployment 

func (c *PostgresBackendRepository) CreateDeployment(ctx context.Context, workspaceId uint, name string, version uint, stubId uint, stubType string, appId uint) (*types.Deployment, error)

func (*PostgresBackendRepository) CreateDiskSnapshot 

func (r *PostgresBackendRepository) CreateDiskSnapshot(ctx context.Context, snapshot *types.DiskSnapshot) (*types.DiskSnapshot, error)

func (*PostgresBackendRepository) CreateImage 

func (r *PostgresBackendRepository) CreateImage(ctx context.Context, imageId string, clipVersion uint32) (uint32, error)

func (*PostgresBackendRepository) CreateObject 

func (r *PostgresBackendRepository) CreateObject(ctx context.Context, hash string, size int64, workspaceId uint) (*types.Object, error)

func (*PostgresBackendRepository) CreateScheduledJob 

func (r *PostgresBackendRepository) CreateScheduledJob(ctx context.Context, scheduledJob *types.ScheduledJob) (*types.ScheduledJob, error)

func (*PostgresBackendRepository) CreateSecret 

func (r *PostgresBackendRepository) CreateSecret(ctx context.Context, workspace *types.Workspace, tokenId uint, name string, value string, validateName bool, hosts []string) (*types.Secret, error)

func (*PostgresBackendRepository) CreateStack 

func (r *PostgresBackendRepository) CreateStack(ctx context.Context, workspaceId uint, name string, spec json.RawMessage) (*types.Stack, error)

func (*PostgresBackendRepository) CreateTask 

func (r *PostgresBackendRepository) CreateTask(ctx context.Context, params *types.TaskParams) (*types.Task, error)

func (*PostgresBackendRepository) CreateToken 

func (r *PostgresBackendRepository) CreateToken(ctx context.Context, workspaceId uint, tokenType string, reusable bool) (types.Token, error)

func (*PostgresBackendRepository) CreateVM 

func (r *PostgresBackendRepository) CreateVM(ctx context.Context, vm *types.VM) error

func (*PostgresBackendRepository) CreateVMArtifact 

func (r *PostgresBackendRepository) CreateVMArtifact(ctx context.Context, ws uint, a *types.VMArtifact) error

func (*PostgresBackendRepository) CreateWorkerToken 

func (r *PostgresBackendRepository) CreateWorkerToken(ctx context.Context, workspaceId uint, tokenType string, workerId string) (types.Token, error)

func (*PostgresBackendRepository) CreateWorkspace 

func (r *PostgresBackendRepository) CreateWorkspace(ctx context.Context) (types.Workspace, error)

func (*PostgresBackendRepository) CreateWorkspaceStorage 

func (r *PostgresBackendRepository) CreateWorkspaceStorage(ctx context.Context, workspaceId uint, storage types.WorkspaceStorage) (*types.WorkspaceStorage, error)

func (*PostgresBackendRepository) DeleteApp 

func (r *PostgresBackendRepository) DeleteApp(ctx context.Context, appId string) error

func (*PostgresBackendRepository) DeleteConcurrencyLimit 

func (r *PostgresBackendRepository) DeleteConcurrencyLimit(ctx context.Context, workspaceId types.Workspace) error

func (*PostgresBackendRepository) DeleteDeployment 

func (r *PostgresBackendRepository) DeleteDeployment(ctx context.Context, deployment types.Deployment) error

func (*PostgresBackendRepository) DeleteDisk 

func (r *PostgresBackendRepository) DeleteDisk(ctx context.Context, workspaceId uint, name string) error

func (*PostgresBackendRepository) DeleteObjectByExternalId 

func (r *PostgresBackendRepository) DeleteObjectByExternalId(ctx context.Context, externalId string) error

func (*PostgresBackendRepository) DeletePreviousScheduledJob 

func (r *PostgresBackendRepository) DeletePreviousScheduledJob(ctx context.Context, deployment *types.Deployment) error

func (*PostgresBackendRepository) DeleteScheduledJob 

func (r *PostgresBackendRepository) DeleteScheduledJob(ctx context.Context, scheduledJob *types.ScheduledJob) error

func (*PostgresBackendRepository) DeleteSecret 

func (r *PostgresBackendRepository) DeleteSecret(ctx context.Context, workspace *types.Workspace, name string) error

func (*PostgresBackendRepository) DeleteStack 

func (r *PostgresBackendRepository) DeleteStack(ctx context.Context, workspaceId uint, externalId string) error

func (*PostgresBackendRepository) DeleteTask 

func (r *PostgresBackendRepository) DeleteTask(ctx context.Context, externalId string) error

func (*PostgresBackendRepository) DeleteToken 

func (r *PostgresBackendRepository) DeleteToken(ctx context.Context, workspaceId uint, extTokenId string) error

func (*PostgresBackendRepository) DeleteVMArtifact 

func (r *PostgresBackendRepository) DeleteVMArtifact(ctx context.Context, ws uint, kind, name string) error

func (*PostgresBackendRepository) DeleteVolume 

func (c *PostgresBackendRepository) DeleteVolume(ctx context.Context, workspaceId uint, name string) error

func (*PostgresBackendRepository) EncryptStoredCredentials 

func (r *PostgresBackendRepository) EncryptStoredCredentials(ctx context.Context) error

EncryptStoredCredentials encrypts token keys and workspace signing keys still stored in plaintext.

func (*PostgresBackendRepository) GetAdminWorkspace 

func (r *PostgresBackendRepository) GetAdminWorkspace(ctx context.Context) (*types.Workspace, error)

func (*PostgresBackendRepository) GetAnyDeploymentByExternalId 

func (c *PostgresBackendRepository) GetAnyDeploymentByExternalId(ctx context.Context, deploymentExternalId string) (*types.DeploymentWithRelated, error)

func (*PostgresBackendRepository) GetCheckpointById 

func (r *PostgresBackendRepository) GetCheckpointById(ctx context.Context, checkpointId string) (*types.Checkpoint, error)

func (*PostgresBackendRepository) GetConcurrencyLimit 

func (r *PostgresBackendRepository) GetConcurrencyLimit(ctx context.Context, concurrencyLimitId uint) (*types.ConcurrencyLimit, error)

func (*PostgresBackendRepository) GetConcurrencyLimitByWorkspaceId 

func (r *PostgresBackendRepository) GetConcurrencyLimitByWorkspaceId(ctx context.Context, workspaceId string) (*types.ConcurrencyLimit, error)

func (*PostgresBackendRepository) GetDeploymentByExternalId 

func (c *PostgresBackendRepository) GetDeploymentByExternalId(ctx context.Context, workspaceId uint, deploymentExternalId string) (*types.DeploymentWithRelated, error)

func (*PostgresBackendRepository) GetDeploymentByNameAndVersion 

func (c *PostgresBackendRepository) GetDeploymentByNameAndVersion(ctx context.Context, workspaceId uint, name string, version uint, stubType string) (*types.DeploymentWithRelated, error)

func (*PostgresBackendRepository) GetDeploymentByStubExternalId 

func (c *PostgresBackendRepository) GetDeploymentByStubExternalId(ctx context.Context, workspaceId uint, stubExternalId string) (*types.DeploymentWithRelated, error)

func (*PostgresBackendRepository) GetDeploymentBySubdomain 

func (c *PostgresBackendRepository) GetDeploymentBySubdomain(ctx context.Context, subdomain string, version uint) (*types.DeploymentWithRelated, error)

GetDeploymentBySubdomain retrieves the deployment by name, version, and stub group If version is 0, it will return the latest version

func (*PostgresBackendRepository) GetDisk 

func (r *PostgresBackendRepository) GetDisk(ctx context.Context, workspaceId uint, name string) (*types.Disk, error)

func (*PostgresBackendRepository) GetDiskSnapshot 

func (r *PostgresBackendRepository) GetDiskSnapshot(ctx context.Context, workspaceId uint, snapshotId string) (*types.DiskSnapshot, error)

func (*PostgresBackendRepository) GetImageClipVersion 

func (r *PostgresBackendRepository) GetImageClipVersion(ctx context.Context, imageId string) (uint32, error)

func (*PostgresBackendRepository) GetImageCredentials 

func (r *PostgresBackendRepository) GetImageCredentials(ctx context.Context, workspace *types.Workspace, imageId string) (string, error)

GetImageCredentials returns the registry credentials for an image, read from the workspace secrets it names, in the form the worker parses. Empty when the image names none.

func (*PostgresBackendRepository) GetLatestCheckpointByStubId 

func (r *PostgresBackendRepository) GetLatestCheckpointByStubId(ctx context.Context, stubExternalId string, compatibilityKeys ...string) (*types.Checkpoint, error)

func (*PostgresBackendRepository) GetLatestDeploymentByName 

func (c *PostgresBackendRepository) GetLatestDeploymentByName(ctx context.Context, workspaceId uint, name string, stubType string, filterDeleted bool) (*types.DeploymentWithRelated, error)

func (*PostgresBackendRepository) GetLatestDiskSnapshot 

func (r *PostgresBackendRepository) GetLatestDiskSnapshot(ctx context.Context, workspaceId uint, diskName string) (*types.DiskSnapshot, error)

func (*PostgresBackendRepository) GetObjectByExternalId 

func (r *PostgresBackendRepository) GetObjectByExternalId(ctx context.Context, externalId string, workspaceId uint) (types.Object, error)

func (*PostgresBackendRepository) GetObjectByExternalStubId 

func (r *PostgresBackendRepository) GetObjectByExternalStubId(ctx context.Context, stubId string, workspaceId uint) (types.Object, error)

func (*PostgresBackendRepository) GetObjectByHash 

func (r *PostgresBackendRepository) GetObjectByHash(ctx context.Context, hash string, workspaceId uint) (*types.Object, error)

func (*PostgresBackendRepository) GetOrCreateApp 

func (r *PostgresBackendRepository) GetOrCreateApp(ctx context.Context, workspaceId uint, appName string) (*types.App, error)

func (*PostgresBackendRepository) GetOrCreateDisk 

func (r *PostgresBackendRepository) GetOrCreateDisk(ctx context.Context, workspaceId uint, disk *types.Disk) (*types.Disk, error)

func (*PostgresBackendRepository) GetOrCreateStub 

func (r *PostgresBackendRepository) GetOrCreateStub(ctx context.Context, name, stubType string, config types.StubConfigV1, objectId, workspaceId uint, forceCreate bool, appId uint) (types.Stub, error)

func (*PostgresBackendRepository) GetOrCreateVolume 

func (c *PostgresBackendRepository) GetOrCreateVolume(ctx context.Context, workspaceId uint, name string) (*types.Volume, error)

func (*PostgresBackendRepository) GetScheduledJob 

func (r *PostgresBackendRepository) GetScheduledJob(ctx context.Context, deploymentId uint) (*types.ScheduledJob, error)

func (*PostgresBackendRepository) GetSecretByName 

func (r *PostgresBackendRepository) GetSecretByName(ctx context.Context, workspace *types.Workspace, name string) (*types.Secret, error)

func (*PostgresBackendRepository) GetSecretByNameDecrypted 

func (r *PostgresBackendRepository) GetSecretByNameDecrypted(ctx context.Context, workspace *types.Workspace, name string) (*types.Secret, error)

func (*PostgresBackendRepository) GetSecretsByName 

func (r *PostgresBackendRepository) GetSecretsByName(ctx context.Context, workspace *types.Workspace, names []string) ([]types.Secret, error)

func (*PostgresBackendRepository) GetSecretsByNameDecrypted 

func (r *PostgresBackendRepository) GetSecretsByNameDecrypted(ctx context.Context, workspace *types.Workspace, names []string) ([]types.Secret, error)

func (*PostgresBackendRepository) GetStubByExternalId 

func (r *PostgresBackendRepository) GetStubByExternalId(ctx context.Context, externalId string, queryFilters ...types.QueryFilter) (*types.StubWithRelated, error)

func (*PostgresBackendRepository) GetStubById 

func (r *PostgresBackendRepository) GetStubById(ctx context.Context, stubId uint, queryFilters ...types.QueryFilter) (*types.StubWithRelated, error)

func (*PostgresBackendRepository) GetTask 

func (r *PostgresBackendRepository) GetTask(ctx context.Context, externalId string) (*types.Task, error)

func (*PostgresBackendRepository) GetTaskByWorkspace 

func (r *PostgresBackendRepository) GetTaskByWorkspace(ctx context.Context, externalId string, workspace *types.Workspace) (*types.TaskWithRelated, error)

func (*PostgresBackendRepository) GetTaskCountPerDeployment 

func (c *PostgresBackendRepository) GetTaskCountPerDeployment(ctx context.Context, filters types.TaskFilter) ([]types.TaskCountPerDeployment, error)

func (*PostgresBackendRepository) GetTaskMetrics 

func (r *PostgresBackendRepository) GetTaskMetrics(ctx context.Context, periodStart, periodEnd time.Time) (types.TaskMetrics, error)

func (*PostgresBackendRepository) GetTaskStatus 

func (r *PostgresBackendRepository) GetTaskStatus(ctx context.Context, externalId string) (types.TaskStatus, error)

func (*PostgresBackendRepository) GetTaskWithRelated 

func (r *PostgresBackendRepository) GetTaskWithRelated(ctx context.Context, externalId string) (*types.TaskWithRelated, error)

func (*PostgresBackendRepository) GetTokenByExternalId 

func (r *PostgresBackendRepository) GetTokenByExternalId(ctx context.Context, workspaceId uint, extTokenId string) (*types.Token, error)

func (*PostgresBackendRepository) GetVM 

func (r *PostgresBackendRepository) GetVM(ctx context.Context, ws uint, name string) (*types.VM, error)

func (*PostgresBackendRepository) GetVMArtifact 

func (r *PostgresBackendRepository) GetVMArtifact(ctx context.Context, ws uint, kind, name string) (*types.VMArtifact, error)

func (*PostgresBackendRepository) GetVMByHandle 

func (r *PostgresBackendRepository) GetVMByHandle(ctx context.Context, handle string) (*types.VM, error)

func (*PostgresBackendRepository) GetVMByPublicIPv6 

func (r *PostgresBackendRepository) GetVMByPublicIPv6(ctx context.Context, address string) (*types.VM, error)

func (*PostgresBackendRepository) GetVolume 

func (c *PostgresBackendRepository) GetVolume(ctx context.Context, workspaceId uint, name string) (*types.Volume, error)

Volume

func (*PostgresBackendRepository) GetVolumeByExternalId 

func (c *PostgresBackendRepository) GetVolumeByExternalId(ctx context.Context, workspaceId uint, externalId string) (*types.Volume, error)

func (*PostgresBackendRepository) GetWorkspace 

func (r *PostgresBackendRepository) GetWorkspace(ctx context.Context, workspaceId uint) (*types.Workspace, error)

func (*PostgresBackendRepository) GetWorkspaceByExternalId 

func (r *PostgresBackendRepository) GetWorkspaceByExternalId(ctx context.Context, externalId string) (types.Workspace, error)

func (*PostgresBackendRepository) GetWorkspaceByExternalIdWithSigningKey 

func (r *PostgresBackendRepository) GetWorkspaceByExternalIdWithSigningKey(ctx context.Context, externalId string) (types.Workspace, error)

func (*PostgresBackendRepository) GetWorkspaceStorage 

func (r *PostgresBackendRepository) GetWorkspaceStorage(ctx context.Context, storageId uint) (*types.WorkspaceStorage, error)

func (*PostgresBackendRepository) ListAppIDsByStubExternalIDs 

func (c *PostgresBackendRepository) ListAppIDsByStubExternalIDs(ctx context.Context, workspaceID string, stubExternalIDs []string) (map[string]string, error)

func (*PostgresBackendRepository) ListApps 

func (r *PostgresBackendRepository) ListApps(ctx context.Context, workspaceId uint) ([]types.App, error)

func (*PostgresBackendRepository) ListAppsPaginated 

func (r *PostgresBackendRepository) ListAppsPaginated(ctx context.Context, workspaceId uint, filters types.AppFilter) (common.CursorPaginationInfo[types.App], error)

func (*PostgresBackendRepository) ListCheckpoints 

func (r *PostgresBackendRepository) ListCheckpoints(ctx context.Context, workspaceExternalId string) ([]types.Checkpoint, error)

func (*PostgresBackendRepository) ListConcurrencyLimitsByWorkspaceId 

func (r *PostgresBackendRepository) ListConcurrencyLimitsByWorkspaceId(ctx context.Context, workspaceId string) ([]types.ConcurrencyLimit, error)

func (*PostgresBackendRepository) ListDeploymentsPaginated 

func (c *PostgresBackendRepository) ListDeploymentsPaginated(ctx context.Context, filters types.DeploymentFilter) (common.CursorPaginationInfo[types.DeploymentWithRelated], error)

func (*PostgresBackendRepository) ListDeploymentsWithRelated 

func (c *PostgresBackendRepository) ListDeploymentsWithRelated(ctx context.Context, filters types.DeploymentFilter) ([]types.DeploymentWithRelated, error)

func (*PostgresBackendRepository) ListDiskSnapshots 

func (r *PostgresBackendRepository) ListDiskSnapshots(ctx context.Context, filter types.DiskSnapshotFilter) ([]types.DiskSnapshot, error)

func (*PostgresBackendRepository) ListDisksWithRelated 

func (r *PostgresBackendRepository) ListDisksWithRelated(ctx context.Context, workspaceId uint) ([]types.DiskWithRelated, error)

func (*PostgresBackendRepository) ListLatestDeploymentsByAppIDs 

func (c *PostgresBackendRepository) ListLatestDeploymentsByAppIDs(ctx context.Context, workspaceID uint, appExternalIDs []string) (map[string]types.DeploymentWithRelated, error)

func (*PostgresBackendRepository) ListLatestDeploymentsWithRelatedPaginated 

func (c *PostgresBackendRepository) ListLatestDeploymentsWithRelatedPaginated(ctx context.Context, filters types.DeploymentFilter) (common.CursorPaginationInfo[types.DeploymentWithRelated], error)

func (*PostgresBackendRepository) ListLatestStubsByAppIDs 

func (c *PostgresBackendRepository) ListLatestStubsByAppIDs(ctx context.Context, workspaceID uint, appExternalIDs []string) (map[string]types.StubWithRelated, error)

func (*PostgresBackendRepository) ListRetiredVMCacheStubKeys 

func (r *PostgresBackendRepository) ListRetiredVMCacheStubKeys(ctx context.Context, keys []string) ([]string, error)

func (*PostgresBackendRepository) ListSecrets 

func (r *PostgresBackendRepository) ListSecrets(ctx context.Context, workspace *types.Workspace) ([]types.Secret, error)

func (*PostgresBackendRepository) ListStacks 

func (r *PostgresBackendRepository) ListStacks(ctx context.Context, workspaceId uint) ([]types.Stack, error)

func (*PostgresBackendRepository) ListStaleCheckpoints 

func (r *PostgresBackendRepository) ListStaleCheckpoints(ctx context.Context, activeRecentStubKeys []string, stubLastUsedBefore time.Time) ([]types.Checkpoint, error)

func (*PostgresBackendRepository) ListStubs 

func (c *PostgresBackendRepository) ListStubs(ctx context.Context, filters types.StubFilter) ([]types.StubWithRelated, error)

func (*PostgresBackendRepository) ListStubsPaginated 

func (c *PostgresBackendRepository) ListStubsPaginated(ctx context.Context, filters types.StubFilter) (common.CursorPaginationInfo[types.StubWithRelated], error)

func (*PostgresBackendRepository) ListTasks 

func (r *PostgresBackendRepository) ListTasks(ctx context.Context) ([]types.Task, error)

func (*PostgresBackendRepository) ListTasksWithRelated 

func (c *PostgresBackendRepository) ListTasksWithRelated(ctx context.Context, filters types.TaskFilter) ([]types.TaskWithRelated, error)

func (*PostgresBackendRepository) ListTasksWithRelatedPaginated 

func (c *PostgresBackendRepository) ListTasksWithRelatedPaginated(ctx context.Context, filters types.TaskFilter) (common.CursorPaginationInfo[types.TaskWithRelated], error)

func (*PostgresBackendRepository) ListTokens 

func (r *PostgresBackendRepository) ListTokens(ctx context.Context, workspaceId uint) ([]types.Token, error)

func (*PostgresBackendRepository) ListVMArtifactPage 

func (r *PostgresBackendRepository) ListVMArtifactPage(ctx context.Context, ws uint, kind string, filter types.VMArtifactFilter) ([]types.VMArtifact, int, error)

func (*PostgresBackendRepository) ListVMArtifacts 

func (r *PostgresBackendRepository) ListVMArtifacts(ctx context.Context, ws uint, kind string) ([]types.VMArtifact, error)

func (*PostgresBackendRepository) ListVMPage 

func (r *PostgresBackendRepository) ListVMPage(ctx context.Context, ws uint, filter types.VMFilter) (*types.VMPage, error)

func (*PostgresBackendRepository) ListVMs 

func (r *PostgresBackendRepository) ListVMs(ctx context.Context, ws uint) ([]*types.VM, error)

func (*PostgresBackendRepository) ListVMsToReconcile 

func (r *PostgresBackendRepository) ListVMsToReconcile(ctx context.Context, backfillIPv6 bool) ([]*types.VM, error)

Keep settled history out of the reconciler's recurring database reads. Stopped/paused VMs are candidates only while public IPv6 needs backfilling.

func (*PostgresBackendRepository) ListVolumesWithRelated 

func (c *PostgresBackendRepository) ListVolumesWithRelated(ctx context.Context, workspaceId uint) ([]types.VolumeWithRelated, error)

func (*PostgresBackendRepository) ListWorkspaces 

func (r *PostgresBackendRepository) ListWorkspaces(ctx context.Context) ([]types.Workspace, error)

func (*PostgresBackendRepository) ListenToChannel 

func (r *PostgresBackendRepository) ListenToChannel(ctx context.Context, channel string) (<-chan string, error)

func (*PostgresBackendRepository) LockVM 

func (r *PostgresBackendRepository) LockVM(ctx context.Context, id string) (func(), error)

LockVM queues user operations behind an existing lifecycle operation without holding a connection or blocking other VMs on the shared lock session.

func (*PostgresBackendRepository) Migrate 

func (r *PostgresBackendRepository) Migrate() error

func (*PostgresBackendRepository) MigrateContext 

func (r *PostgresBackendRepository) MigrateContext(ctx context.Context) error

func (*PostgresBackendRepository) Ping 

func (r *PostgresBackendRepository) Ping() error

func (*PostgresBackendRepository) PruneCheckpoints 

func (r *PostgresBackendRepository) PruneCheckpoints(ctx context.Context, checkpointIds []string) ([]types.Checkpoint, error)

func (*PostgresBackendRepository) PublishDiskSnapshot 

func (r *PostgresBackendRepository) PublishDiskSnapshot(ctx context.Context, workspaceID uint, snapshotID string) error

PublishDiskSnapshot atomically publishes a complete immutable restore chain.

func (*PostgresBackendRepository) RetrieveActiveToken 

func (r *PostgresBackendRepository) RetrieveActiveToken(ctx context.Context, workspaceId uint, tokenType string) (*types.Token, error)

func (*PostgresBackendRepository) RetrieveApp 

func (r *PostgresBackendRepository) RetrieveApp(ctx context.Context, workspaceId uint, appId string) (*types.App, error)

func (*PostgresBackendRepository) RetrieveAppByStubExternalId 

func (r *PostgresBackendRepository) RetrieveAppByStubExternalId(ctx context.Context, stubExternalId string) (*types.App, error)

func (*PostgresBackendRepository) RevertToConcurrencyLimit 

func (r *PostgresBackendRepository) RevertToConcurrencyLimit(ctx context.Context, workspaceId string, concurrencyLimitId string) (*types.ConcurrencyLimit, error)

func (*PostgresBackendRepository) RevokeTokenByExternalId 

func (r *PostgresBackendRepository) RevokeTokenByExternalId(ctx context.Context, externalId string) (string, error)

RevokeTokenByExternalId deactivates a token and returns its key, which is empty when there is no such token.

func (*PostgresBackendRepository) SaveVM 

func (r *PostgresBackendRepository) SaveVM(ctx context.Context, vm *types.VM) error

func (*PostgresBackendRepository) SetImageCredentialSecretNames 

func (r *PostgresBackendRepository) SetImageCredentialSecretNames(ctx context.Context, imageId string, names []string) error

func (*PostgresBackendRepository) SetWorkspaceTokensDisabled 

func (r *PostgresBackendRepository) SetWorkspaceTokensDisabled(ctx context.Context, workspaceId uint, disabled bool) ([]string, error)

SetWorkspaceTokensDisabled sets disabled_by_cluster_admin on every token in the workspace except cluster admin and cluster worker tokens, and returns their keys.

func (*PostgresBackendRepository) ToggleToken 

func (r *PostgresBackendRepository) ToggleToken(ctx context.Context, workspaceId uint, extTokenId string) (types.Token, error)

func (*PostgresBackendRepository) TouchVM 

func (r *PostgresBackendRepository) TouchVM(ctx context.Context, id string) error

func (*PostgresBackendRepository) TryLockVM 

func (r *PostgresBackendRepository) TryLockVM(ctx context.Context, id string) (func(), error)

TryLockVM lets background reconciliation skip busy VMs. Session advisory locks fence lifecycle operations across gateway replicas. Sessions close when their last lock is released; PostgreSQL also releases all their locks when a gateway dies. Persisted intent remains recoverable.

func (*PostgresBackendRepository) UpdateCheckpoint 

func (r *PostgresBackendRepository) UpdateCheckpoint(ctx context.Context, checkpoint *types.Checkpoint) (*types.Checkpoint, error)

func (*PostgresBackendRepository) UpdateConcurrencyLimit 

func (r *PostgresBackendRepository) UpdateConcurrencyLimit(ctx context.Context, concurrencyLimitId uint, gpuLimit uint32, cpuMillicoreLimit uint32) (*types.ConcurrencyLimit, error)

func (*PostgresBackendRepository) UpdateDeployment 

func (r *PostgresBackendRepository) UpdateDeployment(ctx context.Context, deployment types.Deployment) (*types.Deployment, error)

func (*PostgresBackendRepository) UpdateDiskSnapshot 

func (r *PostgresBackendRepository) UpdateDiskSnapshot(ctx context.Context, snapshot *types.DiskSnapshot) (*types.DiskSnapshot, error)

func (*PostgresBackendRepository) UpdateObjectSizeByExternalId 

func (r *PostgresBackendRepository) UpdateObjectSizeByExternalId(ctx context.Context, externalId string, size int) error

func (*PostgresBackendRepository) UpdateSecret 

func (r *PostgresBackendRepository) UpdateSecret(ctx context.Context, workspace *types.Workspace, tokenId uint, secretName string, value string, hosts []string) (*types.Secret, error)

func (*PostgresBackendRepository) UpdateStack 

func (r *PostgresBackendRepository) UpdateStack(ctx context.Context, workspaceId uint, externalId, name string, spec json.RawMessage) (*types.Stack, error)

func (*PostgresBackendRepository) UpdateStackIfUnchanged 

func (r *PostgresBackendRepository) UpdateStackIfUnchanged(ctx context.Context, workspaceId uint, externalId, name string, previous, spec json.RawMessage) (*types.Stack, error)

UpdateStackIfUnchanged serializes agent checkpoints across gateway replicas. Comparing the previous JSON also detects concurrent dashboard edits.

func (*PostgresBackendRepository) UpdateStubConfig 

func (r *PostgresBackendRepository) UpdateStubConfig(ctx context.Context, stubId uint, config *types.StubConfigV1) error

func (*PostgresBackendRepository) UpdateTask 

func (r *PostgresBackendRepository) UpdateTask(ctx context.Context, externalId string, updatedTask types.Task) (*types.Task, error)

type PublicVMRepository 

type PublicVMRepository interface {
	GetVMByPublicIPv6(context.Context, string) (*types.VM, error)
	AssignVMPublicIPv6(context.Context, string, string) (string, error)
}

PublicVMRepository is required only when public VM networking is enabled.

type RedisCommandCounts 

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

RedisCommandCounts counts what a test Redis is asked to do between Start and Stop: every command the server runs, the ones a script issues among them (as Redis's commandstats counts them), and the round trips the client makes. A pipeline or a script is one round trip.

func NewRedisClientForTestWithCounts 

func NewRedisClientForTestWithCounts() (*common.RedisClient, *RedisCommandCounts, error)

NewRedisClientForTestWithCounts is NewRedisClientForTest with a counter of the commands and round trips the client's Redis sees.

func (*RedisCommandCounts) DialHook 

func (c *RedisCommandCounts) DialHook(next redis.DialHook) redis.DialHook

func (*RedisCommandCounts) ProcessHook 

func (c *RedisCommandCounts) ProcessHook(next redis.ProcessHook) redis.ProcessHook

func (*RedisCommandCounts) ProcessPipelineHook 

func (c *RedisCommandCounts) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook

func (*RedisCommandCounts) Start 

func (c *RedisCommandCounts) Start()

Start begins a new count.

func (*RedisCommandCounts) Stop 

func (c *RedisCommandCounts) Stop() (map[string]int, int)

Stop ends the count and returns the commands the server ran, by name, and the round trips the client made.

type S2EventRepository 

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

S2EventRepository reads and writes events and logs in one S2 basin.

func NewS2EventRepository 

func NewS2EventRepository(config types.S2Config) (*S2EventRepository, error)

func (*S2EventRepository) GetContainerEvents 

func (r *S2EventRepository) GetContainerEvents(ctx context.Context, containerID string, query types.EventQuery) (*types.ContainerEventsResponse, error)

func (*S2EventRepository) GetEventHistory 

func (r *S2EventRepository) GetEventHistory(ctx context.Context, query types.EventQuery) (*types.EventHistoryResponse, error)

func (*S2EventRepository) GetLogs 

func (r *S2EventRepository) GetLogs(ctx context.Context, query types.LogQuery) (*types.LogsResponse, error)

func (*S2EventRepository) GetPoolMetricsTimeseries 

func (r *S2EventRepository) GetPoolMetricsTimeseries(ctx context.Context, query types.EventQuery, start, end time.Time, interval string) (*types.PoolMetricsTimeseriesResponse, error)

func (*S2EventRepository) GetStubMetricsTimeseries 

func (r *S2EventRepository) GetStubMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.MetricsTimeseriesResponse, error)

func (*S2EventRepository) GetWorkspaceMetricsTimeseries 

func (r *S2EventRepository) GetWorkspaceMetricsTimeseries(ctx context.Context, query types.EventQuery, start time.Time, end time.Time, interval string) (*types.MetricsTimeseriesResponse, error)

func (*S2EventRepository) PushEvent 

func (r *S2EventRepository) PushEvent(event cloudevents.Event) error

func (*S2EventRepository) PushEventSync 

func (r *S2EventRepository) PushEventSync(event cloudevents.Event) error

func (*S2EventRepository) ReadStubCacheRequiredContent 

func (r *S2EventRepository) ReadStubCacheRequiredContent(ctx context.Context, workspaceID, stubID string) ([]types.CacheRequiredContentItem, error)

ReadStubCacheRequiredContent reads the coalesced required-content set for a stub from the dedicated S2 cache stream. Items are merged by immutable content identity or mutable source path, keeping the latest report.

func (*S2EventRepository) StreamAppNamespaceEvents 

func (r *S2EventRepository) StreamAppNamespaceEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

func (*S2EventRepository) StreamContainerEvents 

func (r *S2EventRepository) StreamContainerEvents(ctx context.Context, containerID string, query types.EventQuery) (EventStream, error)

func (*S2EventRepository) StreamLogs 

func (r *S2EventRepository) StreamLogs(ctx context.Context, query types.LogQuery) (EventStream, error)

func (*S2EventRepository) StreamStubEvents 

func (r *S2EventRepository) StreamStubEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

func (*S2EventRepository) StreamTaskEvents 

func (r *S2EventRepository) StreamTaskEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

func (*S2EventRepository) StreamWorkspaceEvents 

func (r *S2EventRepository) StreamWorkspaceEvents(ctx context.Context, query types.EventQuery) (EventStream, error)

type SchedulerRepository 

type SchedulerRepository interface{}

type ScopedS2EventRepository 

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

ScopedS2EventRepository writes events with tokens scoped to stream prefixes, each stream through the first target whose prefix holds it.

func NewScopedS2EventRepository 

func NewScopedS2EventRepository(config types.S2Config) (*ScopedS2EventRepository, error)

func (*ScopedS2EventRepository) PushEvent 

func (r *ScopedS2EventRepository) PushEvent(event cloudevents.Event) error

func (*ScopedS2EventRepository) PushEventSync 

func (r *ScopedS2EventRepository) PushEventSync(event cloudevents.Event) error

type TaskRedisRepository 

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

func (*TaskRedisRepository) ClaimTask 

func (r *TaskRedisRepository) ClaimTask(ctx context.Context, workspaceName, stubId, taskId, containerId string) error

func (*TaskRedisRepository) DeleteTaskState 

func (r *TaskRedisRepository) DeleteTaskState(ctx context.Context, workspaceName, stubId, taskId string) error

func (*TaskRedisRepository) GetTaskState 

func (r *TaskRedisRepository) GetTaskState(ctx context.Context, workspaceName, stubId, taskId string) (*types.TaskMessage, error)

func (*TaskRedisRepository) GetTasksInFlight 

func (r *TaskRedisRepository) GetTasksInFlight(ctx context.Context) ([]*types.TaskMessage, error)

func (*TaskRedisRepository) IsClaimed 

func (r *TaskRedisRepository) IsClaimed(ctx context.Context, workspaceName, stubId, taskId string) (bool, error)

func (*TaskRedisRepository) RemoveTaskClaim 

func (r *TaskRedisRepository) RemoveTaskClaim(ctx context.Context, workspaceName, stubId, taskId string) error

func (*TaskRedisRepository) SetTaskState 

func (r *TaskRedisRepository) SetTaskState(ctx context.Context, workspaceName, stubId, taskId string, msg []byte) error

func (*TaskRedisRepository) TasksClaimed 

func (r *TaskRedisRepository) TasksClaimed(ctx context.Context, workspaceName, stubId string) (int, error)

func (*TaskRedisRepository) TasksInFlight 

func (r *TaskRedisRepository) TasksInFlight(ctx context.Context, workspaceName, stubId string) (int, error)

func (*TaskRedisRepository) WithTaskMonitorLease 

func (r *TaskRedisRepository) WithTaskMonitorLease(ctx context.Context, fn func(context.Context) error) error

func (*TaskRedisRepository) WithTaskRetryLease 

func (r *TaskRedisRepository) WithTaskRetryLease(ctx context.Context, workspaceName, stubId, taskId string, fn func(context.Context) error) error

type TaskRepository 

type TaskRepository interface {
	WithTaskMonitorLease(ctx context.Context, fn func(context.Context) error) error
	GetTaskState(ctx context.Context, workspaceName, stubId, taskId string) (*types.TaskMessage, error)
	SetTaskState(ctx context.Context, workspaceName, stubId, taskId string, msg []byte) error
	DeleteTaskState(ctx context.Context, workspaceName, stubId, taskId string) error
	GetTasksInFlight(ctx context.Context) ([]*types.TaskMessage, error)
	ClaimTask(ctx context.Context, workspaceName, stubId, taskId, containerId string) error
	RemoveTaskClaim(ctx context.Context, workspaceName, stubId, taskId string) error
	IsClaimed(ctx context.Context, workspaceName, stubId, taskId string) (bool, error)
	TasksClaimed(ctx context.Context, workspaceName, stubId string) (int, error)
	TasksInFlight(ctx context.Context, workspaceName, stubId string) (int, error)
	WithTaskRetryLease(ctx context.Context, workspaceName, stubId, taskId string, fn func(context.Context) error) error
}

func NewTaskRedisRepository 

func NewTaskRedisRepository(r *common.RedisClient) TaskRepository

type ThunderClientEnrollmentState 

type ThunderClientEnrollmentState struct {
	ContainerID       string `json:"container_id"`
	WorkspaceID       string `json:"workspace_id"`
	WorkerID          string `json:"worker_id"`
	MachineID         string `json:"machine_id"`
	PoolName          string `json:"pool_name"`
	EnrollmentTokenID string `json:"enrollment_token_id"`
}

type ThunderNodeEnrollmentState 

type ThunderNodeEnrollmentState struct {
	WorkspaceID       string `json:"workspace_id"`
	PoolName          string `json:"pool_name"`
	MachineID         string `json:"machine_id"`
	EnrollmentTokenID string `json:"enrollment_token_id"`
}

type ThunderRedisRepository 

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

func NewThunderRedisRepository 

func NewThunderRedisRepository(rdb *common.RedisClient) *ThunderRedisRepository

func (*ThunderRedisRepository) DeleteClientEnrollment 

func (r *ThunderRedisRepository) DeleteClientEnrollment(ctx context.Context, containerID string) error

func (*ThunderRedisRepository) DeleteNodeEnrollment 

func (r *ThunderRedisRepository) DeleteNodeEnrollment(ctx context.Context, workspaceID, poolName, machineID string) error

func (*ThunderRedisRepository) DeleteZone 

func (r *ThunderRedisRepository) DeleteZone(ctx context.Context, workspaceID, poolName string) error

func (*ThunderRedisRepository) GetClientEnrollment 

func (r *ThunderRedisRepository) GetClientEnrollment(ctx context.Context, containerID string) (*ThunderClientEnrollmentState, bool, error)

func (*ThunderRedisRepository) GetNodeEnrollment 

func (r *ThunderRedisRepository) GetNodeEnrollment(ctx context.Context, workspaceID, poolName, machineID string) (*ThunderNodeEnrollmentState, bool, error)

func (*ThunderRedisRepository) GetZone 

func (r *ThunderRedisRepository) GetZone(ctx context.Context, workspaceID, poolName string) (*ThunderZoneState, bool, error)

func (*ThunderRedisRepository) ListClientEnrollments 

func (r *ThunderRedisRepository) ListClientEnrollments(ctx context.Context) ([]*ThunderClientEnrollmentState, error)

func (*ThunderRedisRepository) ListNodeEnrollments 

func (r *ThunderRedisRepository) ListNodeEnrollments(ctx context.Context, workspaceID, poolName string) ([]*ThunderNodeEnrollmentState, error)

func (*ThunderRedisRepository) ListZones 

func (r *ThunderRedisRepository) ListZones(ctx context.Context, workspaceID string) ([]*ThunderZoneState, error)

func (*ThunderRedisRepository) SaveClientEnrollment 

func (r *ThunderRedisRepository) SaveClientEnrollment(ctx context.Context, state *ThunderClientEnrollmentState) error

func (*ThunderRedisRepository) SaveNodeEnrollment 

func (r *ThunderRedisRepository) SaveNodeEnrollment(ctx context.Context, state *ThunderNodeEnrollmentState) error

func (*ThunderRedisRepository) SaveZone 

func (r *ThunderRedisRepository) SaveZone(ctx context.Context, state *ThunderZoneState) error

func (*ThunderRedisRepository) WithPoolLock 

func (r *ThunderRedisRepository) WithPoolLock(ctx context.Context, workspaceID, poolName string, fn func(context.Context) error) error

type ThunderRepository 

type ThunderRepository interface {
	WithPoolLock(ctx context.Context, workspaceID, poolName string, fn func(context.Context) error) error

	GetClientEnrollment(ctx context.Context, containerID string) (*ThunderClientEnrollmentState, bool, error)
	SaveClientEnrollment(ctx context.Context, state *ThunderClientEnrollmentState) error
	DeleteClientEnrollment(ctx context.Context, containerID string) error
	ListClientEnrollments(ctx context.Context) ([]*ThunderClientEnrollmentState, error)

	GetNodeEnrollment(ctx context.Context, workspaceID, poolName, machineID string) (*ThunderNodeEnrollmentState, bool, error)
	SaveNodeEnrollment(ctx context.Context, state *ThunderNodeEnrollmentState) error
	DeleteNodeEnrollment(ctx context.Context, workspaceID, poolName, machineID string) error
	ListNodeEnrollments(ctx context.Context, workspaceID, poolName string) ([]*ThunderNodeEnrollmentState, error)

	GetZone(ctx context.Context, workspaceID, poolName string) (*ThunderZoneState, bool, error)
	SaveZone(ctx context.Context, state *ThunderZoneState) error
	DeleteZone(ctx context.Context, workspaceID, poolName string) error
	ListZones(ctx context.Context, workspaceID string) ([]*ThunderZoneState, error)
}

type ThunderZoneState 

type ThunderZoneState struct {
	WorkspaceID   string `json:"workspace_id"`
	PoolName      string `json:"pool_name"`
	ThunderZoneID string `json:"thunder_zone_id"`
}

type UsageMeterReader 

type UsageMeterReader interface {
	QueryMeter(ctx context.Context, meter, subject string, from, to time.Time, groupBy []string) ([]types.MeterRow, error)
}

UsageMeterReader is the query side of a usage collector that keeps meters: one aggregate per UTC day for a subject, split by the group-by dimensions. A collector that only exports counters does not implement it.

type UsageMetricsRepository 

type UsageMetricsRepository interface {
	Init(source string) error
	IncrementCounter(name string, metadata map[string]interface{}, value float64) error
	SetGauge(name string, metadata map[string]interface{}, value float64) error
}

type VMRepository 

type VMRepository interface {
	CreateVM(context.Context, *types.VM) error
	SaveVM(context.Context, *types.VM) error
	GetVM(context.Context, uint, string) (*types.VM, error)
	GetVMByHandle(context.Context, string) (*types.VM, error)
	ListVMs(context.Context, uint) ([]*types.VM, error)
	ListVMPage(context.Context, uint, types.VMFilter) (*types.VMPage, error)
	ListVMsToReconcile(context.Context, bool) ([]*types.VM, error)
	LockVM(context.Context, string) (func(), error)
	TryLockVM(context.Context, string) (func(), error)
	TouchVM(context.Context, string) error
	ClaimVMIdleStop(context.Context, string, time.Time) (bool, error)
	CreateVMArtifact(context.Context, uint, *types.VMArtifact) error
	GetVMArtifact(context.Context, uint, string, string) (*types.VMArtifact, error)
	ListVMArtifacts(context.Context, uint, string) ([]types.VMArtifact, error)
	ListVMArtifactPage(context.Context, uint, string, types.VMArtifactFilter) ([]types.VMArtifact, int, error)
	DeleteVMArtifact(context.Context, uint, string, string) error
}

type WebhookSource 

type WebhookSource interface {
	ListWebhooks(ctx context.Context, workspaceId string) ([]types.WorkspaceWebhook, error)
}

WebhookSource lists a workspace's webhooks; the sink caches results briefly.

type WorkerPoolRedisRepository 

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

func (*WorkerPoolRedisRepository) DeleteWorkerPoolState 

func (r *WorkerPoolRedisRepository) DeleteWorkerPoolState(ctx context.Context, poolName string) error

DeleteWorkerPoolState removes the cached health snapshot for a pool. Pool configuration lives elsewhere, so this is safe to call while deleting a API-managed pool and prevents a later same-name pool from inheriting stale health or capacity data.

func (*WorkerPoolRedisRepository) GetWorkerPoolState 

func (r *WorkerPoolRedisRepository) GetWorkerPoolState(ctx context.Context, poolName string) (*types.WorkerPoolState, error)

GetWorkerPoolState retrieves a collection of health metrics for a worker pool

func (*WorkerPoolRedisRepository) RemoveWorkerCleanerLock 

func (r *WorkerPoolRedisRepository) RemoveWorkerCleanerLock(poolName string) error

func (*WorkerPoolRedisRepository) RemoveWorkerPoolSizerLock 

func (r *WorkerPoolRedisRepository) RemoveWorkerPoolSizerLock(poolName string) error

func (*WorkerPoolRedisRepository) RemoveWorkerPoolStateLock 

func (r *WorkerPoolRedisRepository) RemoveWorkerPoolStateLock(poolName string) error

func (*WorkerPoolRedisRepository) SetWorkerCleanerLock 

func (r *WorkerPoolRedisRepository) SetWorkerCleanerLock(poolName string) error

func (*WorkerPoolRedisRepository) SetWorkerPoolSizerLock 

func (r *WorkerPoolRedisRepository) SetWorkerPoolSizerLock(poolName string) error

func (*WorkerPoolRedisRepository) SetWorkerPoolState 

func (r *WorkerPoolRedisRepository) SetWorkerPoolState(ctx context.Context, poolName string, state *types.WorkerPoolState) error

SetWorkerPoolState updates the worker pool state with some recent health metrics

func (*WorkerPoolRedisRepository) SetWorkerPoolStateLock 

func (r *WorkerPoolRedisRepository) SetWorkerPoolStateLock(poolName string) error

type WorkerPoolRepository 

type WorkerPoolRepository interface {
	SetWorkerPoolState(ctx context.Context, poolName string, state *types.WorkerPoolState) error
	GetWorkerPoolState(ctx context.Context, poolName string) (*types.WorkerPoolState, error)
	DeleteWorkerPoolState(ctx context.Context, poolName string) error
	SetWorkerPoolStateLock(poolName string) error
	RemoveWorkerPoolStateLock(poolName string) error
	SetWorkerPoolSizerLock(poolName string) error
	RemoveWorkerPoolSizerLock(poolName string) error
	SetWorkerCleanerLock(poolName string) error
	RemoveWorkerCleanerLock(poolName string) error
}

func NewWorkerPoolRedisRepository 

func NewWorkerPoolRedisRepository(rdb *common.RedisClient) WorkerPoolRepository

func NewWorkerPoolRedisRepositoryForTest 

func NewWorkerPoolRedisRepositoryForTest(rdb *common.RedisClient) WorkerPoolRepository

type WorkerRedisRepository 

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

func (*WorkerRedisRepository) AddContainerToWorker 

func (r *WorkerRedisRepository) AddContainerToWorker(workerId, containerId, deliveryToken string) error

func (*WorkerRedisRepository) AddWorker 

func (r *WorkerRedisRepository) AddWorker(worker *types.Worker) error

AddWorker adds or updates a worker

func (*WorkerRedisRepository) CancelWorkerRollout 

func (r *WorkerRedisRepository) CancelWorkerRollout(workerId, generation string) error

CancelWorkerRollout restores an unpublished rollout's original readiness. The caller holds the agent slot lock and has verified that the published slot still matches the desired generation. Compare the pending target under the scheduling lock so a stale snapshot cannot cancel another transition.

func (*WorkerRedisRepository) GetAllWorkers 

func (r *WorkerRedisRepository) GetAllWorkers() ([]*types.Worker, error)

func (*WorkerRedisRepository) GetAllWorkersInPool 

func (r *WorkerRedisRepository) GetAllWorkersInPool(poolName string) ([]*types.Worker, error)

func (*WorkerRedisRepository) GetContainerIp 

func (r *WorkerRedisRepository) GetContainerIp(networkPrefix string, containerId string) (string, error)

func (*WorkerRedisRepository) GetContainerIpAssignments 

func (r *WorkerRedisRepository) GetContainerIpAssignments(networkPrefix string) ([]types.ContainerIpAssignment, error)

func (*WorkerRedisRepository) GetContainerIps 

func (r *WorkerRedisRepository) GetContainerIps(networkPrefix string) ([]string, error)

func (*WorkerRedisRepository) GetFreeGpuCounts 

func (r *WorkerRedisRepository) GetFreeGpuCounts() (map[string]int, error)

func (*WorkerRedisRepository) GetGpuAvailability 

func (r *WorkerRedisRepository) GetGpuAvailability() (map[string]bool, error)

func (*WorkerRedisRepository) GetGpuCounts 

func (r *WorkerRedisRepository) GetGpuCounts() (map[string]int, error)

func (*WorkerRedisRepository) GetId 

func (r *WorkerRedisRepository) GetId() string

func (*WorkerRedisRepository) GetNextContainerRequest 

func (r *WorkerRedisRepository) GetNextContainerRequest(workerId string) (*types.ContainerRequest, error)

func (*WorkerRedisRepository) GetNextContainerRequests 

func (r *WorkerRedisRepository) GetNextContainerRequests(workerId string, limit int) ([]*types.ContainerRequest, error)

func (*WorkerRedisRepository) GetPreemptibleGpus 

func (r *WorkerRedisRepository) GetPreemptibleGpus() []string

func (*WorkerRedisRepository) GetWorkerById 

func (r *WorkerRedisRepository) GetWorkerById(workerId string) (*types.Worker, error)

func (*WorkerRedisRepository) MoveContainerIp 

func (r *WorkerRedisRepository) MoveContainerIp(networkPrefix, fromContainerId, toContainerId, containerIp string) error

func (*WorkerRedisRepository) PrepareWorkerRollout 

func (r *WorkerRedisRepository) PrepareWorkerRollout(workerId, generation string) (bool, error)

PrepareWorkerRollout disables an agent worker immediately, then waits until queued, delivered, and active container work is gone before allowing the agent slot to change. The worker lock is shared with scheduling, so no new assignment can race the transition.

func (*WorkerRedisRepository) RecoverPendingContainerRequests 

func (r *WorkerRedisRepository) RecoverPendingContainerRequests(workerId string) error

func (*WorkerRedisRepository) RemoveContainerFromWorker 

func (r *WorkerRedisRepository) RemoveContainerFromWorker(workerId string, containerId string) error

func (*WorkerRedisRepository) RemoveContainerIp 

func (r *WorkerRedisRepository) RemoveContainerIp(networkPrefix string, containerId string) error

func (*WorkerRedisRepository) RemoveImagePullLock 

func (r *WorkerRedisRepository) RemoveImagePullLock(workerId, imageId, token string) error

func (*WorkerRedisRepository) RemoveNetworkLock 

func (r *WorkerRedisRepository) RemoveNetworkLock(networkPrefix string, token string) error

func (*WorkerRedisRepository) RemoveWorker 

func (r *WorkerRedisRepository) RemoveWorker(workerId string) error

func (*WorkerRedisRepository) RemoveWorkerNetworkState 

func (r *WorkerRedisRepository) RemoveWorkerNetworkState(ctx context.Context, networkPrefix string) error

func (*WorkerRedisRepository) RequeueContainerRequests 

func (r *WorkerRedisRepository) RequeueContainerRequests(workerId string, requests []*types.ContainerRequest) error

func (*WorkerRedisRepository) ScheduleContainerRequest 

func (r *WorkerRedisRepository) ScheduleContainerRequest(worker *types.Worker, request *types.ContainerRequest) error

func (*WorkerRedisRepository) ScheduleContainerRequests 

func (r *WorkerRedisRepository) ScheduleContainerRequests(worker *types.Worker, requests []*types.ContainerRequest) error

func (*WorkerRedisRepository) SetContainerIp 

func (r *WorkerRedisRepository) SetContainerIp(networkPrefix string, containerId, containerIp string) error

func (*WorkerRedisRepository) SetContainerResourceValues 

func (r *WorkerRedisRepository) SetContainerResourceValues(workerId string, containerId string, usage types.ContainerResourceUsage) error

func (*WorkerRedisRepository) SetImagePullLock 

func (r *WorkerRedisRepository) SetImagePullLock(workerId, imageId string) (string, error)

func (*WorkerRedisRepository) SetNetworkLock 

func (r *WorkerRedisRepository) SetNetworkLock(networkPrefix string, ttl, retries int) (string, error)

func (*WorkerRedisRepository) SetWorkerCordon 

func (r *WorkerRedisRepository) SetWorkerCordon(workerId string, cordoned bool) error

func (*WorkerRedisRepository) SetWorkerKeepAlive 

func (r *WorkerRedisRepository) SetWorkerKeepAlive(workerId string, keepAlive types.WorkerKeepAlive) error

func (*WorkerRedisRepository) ToggleWorkerAvailable 

func (r *WorkerRedisRepository) ToggleWorkerAvailable(workerId, generation string) error

func (*WorkerRedisRepository) UpdateWorkerCapacity 

func (r *WorkerRedisRepository) UpdateWorkerCapacity(worker *types.Worker, request *types.ContainerRequest, capacityUpdateType types.CapacityUpdateType) error

func (*WorkerRedisRepository) UpdateWorkerStatus 

func (r *WorkerRedisRepository) UpdateWorkerStatus(workerId string, status types.WorkerStatus) error

type WorkerRepository 

type WorkerRepository interface {
	GetId() string
	GetWorkerById(workerId string) (*types.Worker, error)
	GetAllWorkers() ([]*types.Worker, error)
	GetAllWorkersInPool(poolName string) ([]*types.Worker, error)
	AddWorker(w *types.Worker) error
	ToggleWorkerAvailable(workerId, generation string) error
	SetWorkerCordon(workerId string, cordoned bool) error
	PrepareWorkerRollout(workerId, generation string) (bool, error)
	CancelWorkerRollout(workerId, generation string) error
	UpdateWorkerStatus(workerId string, status types.WorkerStatus) error
	RemoveWorker(workerId string) error
	SetWorkerKeepAlive(workerId string, keepAlive types.WorkerKeepAlive) error
	UpdateWorkerCapacity(w *types.Worker, cr *types.ContainerRequest, ut types.CapacityUpdateType) error
	ScheduleContainerRequest(worker *types.Worker, request *types.ContainerRequest) error
	ScheduleContainerRequests(worker *types.Worker, requests []*types.ContainerRequest) error
	GetNextContainerRequest(workerId string) (*types.ContainerRequest, error)
	GetNextContainerRequests(workerId string, limit int) ([]*types.ContainerRequest, error)
	RecoverPendingContainerRequests(workerId string) error
	RequeueContainerRequests(workerId string, requests []*types.ContainerRequest) error
	AddContainerToWorker(workerId, containerId, deliveryToken string) error
	RemoveContainerFromWorker(workerId string, containerId string) error
	SetContainerResourceValues(workerId string, containerId string, usage types.ContainerResourceUsage) error
	SetImagePullLock(workerId, imageId string) (string, error)
	RemoveImagePullLock(workerId, imageId, token string) error
	GetContainerIp(networkPrefix string, containerId string) (string, error)
	SetContainerIp(networkPrefix string, containerId, containerIp string) error
	MoveContainerIp(networkPrefix, fromContainerId, toContainerId, containerIp string) error
	RemoveContainerIp(networkPrefix string, containerId string) error
	GetContainerIps(networkPrefix string) ([]string, error)
	GetContainerIpAssignments(networkPrefix string) ([]types.ContainerIpAssignment, error)
	RemoveWorkerNetworkState(ctx context.Context, networkPrefix string) error
	SetNetworkLock(networkPrefix string, ttl, retries int) (string, error)
	RemoveNetworkLock(networkPrefix string, token string) error
	GetGpuCounts() (map[string]int, error)
	GetGpuAvailability() (map[string]bool, error)
	GetFreeGpuCounts() (map[string]int, error)
	GetPreemptibleGpus() []string
}

func NewWorkerRedisRepository 

func NewWorkerRedisRepository(r *common.RedisClient, config types.WorkerConfig) WorkerRepository

func NewWorkerRedisRepositoryForTest 

func NewWorkerRedisRepositoryForTest(rdb *common.RedisClient) WorkerRepository

type WorkspaceRedisRepository 

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

func (*WorkspaceRedisRepository) AuthorizeToken 

func (wr *WorkspaceRedisRepository) AuthorizeToken(token string) (*types.Token, *types.Workspace, error)

func (*WorkspaceRedisRepository) DeleteWebhook 

func (wr *WorkspaceRedisRepository) DeleteWebhook(ctx context.Context, workspaceId, webhookId string) error

func (*WorkspaceRedisRepository) GetConcurrencyLimitByWorkspaceId 

func (wr *WorkspaceRedisRepository) GetConcurrencyLimitByWorkspaceId(workspaceId string) (*types.ConcurrencyLimit, error)

func (*WorkspaceRedisRepository) ListWebhooks 

func (wr *WorkspaceRedisRepository) ListWebhooks(ctx context.Context, workspaceId string) ([]types.WorkspaceWebhook, error)

Webhooks live in one hash per workspace: field = webhook id, value = JSON.

func (*WorkspaceRedisRepository) RevokeToken 

func (wr *WorkspaceRedisRepository) RevokeToken(tokenKey string) error

func (*WorkspaceRedisRepository) SetAuthorizationToken 

func (wr *WorkspaceRedisRepository) SetAuthorizationToken(token *types.Token, workspace *types.Workspace) error

func (*WorkspaceRedisRepository) SetConcurrencyLimitByWorkspaceId 

func (wr *WorkspaceRedisRepository) SetConcurrencyLimitByWorkspaceId(workspaceId string, limit *types.ConcurrencyLimit) error

func (*WorkspaceRedisRepository) SetWebhook 

func (wr *WorkspaceRedisRepository) SetWebhook(ctx context.Context, workspaceId string, webhook types.WorkspaceWebhook) error

type WorkspaceRepository 

type WorkspaceRepository interface {
	GetConcurrencyLimitByWorkspaceId(workspaceId string) (*types.ConcurrencyLimit, error)
	SetConcurrencyLimitByWorkspaceId(workspaceId string, limit *types.ConcurrencyLimit) error
	ListWebhooks(ctx context.Context, workspaceId string) ([]types.WorkspaceWebhook, error)
	SetWebhook(ctx context.Context, workspaceId string, webhook types.WorkspaceWebhook) error
	DeleteWebhook(ctx context.Context, workspaceId, webhookId string) error
	AuthorizeToken(string) (*types.Token, *types.Workspace, error)
	RevokeToken(tokenKey string) error
	SetAuthorizationToken(*types.Token, *types.Workspace) error
}

func NewWorkspaceRedisRepository 

func NewWorkspaceRedisRepository(r *common.RedisClient, encryptionKey string) WorkspaceRepository

func NewWorkspaceRedisRepositoryForTest 

func NewWorkspaceRedisRepositoryForTest(rdb *common.RedisClient) WorkspaceRepository

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.