v1alpha

type AssignedItemWithTask ¶

type AssignedItemWithTask = scheduling.AssignedItemWithTask

type BatchScheduler ¶

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

func (b *BatchScheduler) Cleanup(ctx context.Context) error

func (b *BatchScheduler) Start(ctx context.Context)

type ConcurrencyManager ¶

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

func (c *ConcurrencyManager) Cleanup()

type ConcurrencyResults ¶

type ConcurrencyResults = scheduling.ConcurrencyResults

type Extensions ¶

type Extensions = scheduling.Extensions

type LabelPairPromLabels ¶

type LabelPairPromLabels struct {
	WorkerLabelPair
	SlotType string
}

LabelPairPromLabels identifies a label-pair slot series: one worker label pair combined with one slot type.

type LeaseManager ¶

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

LeaseManager is responsible for leases on multiple queues and multiplexing queue results to callers. It is still tenant-scoped.

type PostAssignInput ¶

type PostAssignInput = scheduling.PostAssignInput

The extension contract crosses the engine boundary — cloud and OSS register extensions into whichever pool implementation a shard runs — so it is defined once in pkg/scheduling and aliased here.

type PrometheusExtension ¶

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

func NewPrometheusExtension(promGate *prometheus.Gate) *PrometheusExtension

func (p *PrometheusExtension) Cleanup() error

func (p *PrometheusExtension) CleanupTenant(tenantId uuid.UUID) error

func (p *PrometheusExtension) PostAssign(tenantId uuid.UUID, input *PostAssignInput)

func (p *PrometheusExtension) ReportSnapshot(ctx context.Context, tenantId uuid.UUID, input *SnapshotInput)

func (p *PrometheusExtension) SetTenants(tenants []*sqlcv1.Tenant)

type QueueResults ¶

type QueueResults = scheduling.QueueResults

These types cross the engine boundary (channels and optimistic-scheduling results), so they are defined once in pkg/scheduling and aliased here: the engine handles identical types no matter which scheduler implementation a shard runs.

type Queuer ¶

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

func (q *Queuer) Cleanup()

type Scheduler ¶

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

Scheduler is responsible for scheduling steps to workers as efficiently as possible. This is tenant-scoped, so each tenant will have its own scheduler.

All mutable scheduling state is owned by a single run-loop goroutine: every read and write happens inside an op sent to the loop via do(). There are no locks and no lock-ordering rules. Database reads run outside the loop, so assignment is never blocked on I/O; replenish reconciles its reads against assignments that completed in the meantime (see ackedDuringReplenish).

type SchedulerExtension ¶

type SchedulerExtension = scheduling.SchedulerExtension

type SchedulingPool ¶

type SchedulingPool struct {
	Extensions *Extensions
	// contains filtered or unexported fields
}

func NewSchedulingPool(
	repo v1.SchedulerRepository,
	taskRepo v1.TaskRepository,
	outbox pgoutbox.Outbox,
	l *zerolog.Logger,
	singleQueueLimit int,
	schedulerConcurrencyRateLimit int,
	schedulerConcurrencyPollingMinInterval time.Duration,
	schedulerConcurrencyPollingMaxInterval time.Duration,
	schedulerCheckActiveMinInterval time.Duration,
	schedulerCheckActiveMaxInterval time.Duration,
	schedulerAdvisoryLockTimeout time.Duration,
	optimisticSchedulingEnabled bool,
	optimisticSlots int,
	concurrencyInMemoryIndexEnabled bool,
	promGate *prometheus.Gate,
) (*SchedulingPool, func() error, error)

func (p *SchedulingPool) AddExtension(ext scheduling.SchedulerExtension)

func (p *SchedulingPool) GetConcurrencyResultsCh() chan *ConcurrencyResults

func (p *SchedulingPool) GetResultsCh() chan *QueueResults

func (p *SchedulingPool) NotifyConcurrency(ctx context.Context, tenantId uuid.UUID, strategyIds []int64)

func (p *SchedulingPool) NotifyNewConcurrencyStrategy(ctx context.Context, tenantId uuid.UUID, strategyId int64)

func (p *SchedulingPool) NotifyNewQueue(ctx context.Context, tenantId uuid.UUID, queueName string)

func (p *SchedulingPool) NotifyNewWorker(ctx context.Context, tenantId uuid.UUID, workerId uuid.UUID)

func (p *SchedulingPool) NotifyQueues(ctx context.Context, tenantId uuid.UUID, queueNames []string)

func (p *SchedulingPool) Replenish(ctx context.Context, tenantId uuid.UUID)

func (p *SchedulingPool) RunOptimisticScheduling(ctx context.Context, tenantId uuid.UUID, opts []*v1.WorkflowNameTriggerOpts, localWorkerIds map[uuid.UUID]struct{}) (map[uuid.UUID][]*AssignedItemWithTask, []*v1.V1TaskWithPayload, []*v1.DAGWithData, []v1.IdempotencyCollision, error)

func (p *SchedulingPool) RunOptimisticSchedulingFromEvents(ctx context.Context, tenantId uuid.UUID, opts []v1.EventTriggerOpts, localWorkerIds map[uuid.UUID]struct{}) (map[uuid.UUID][]*AssignedItemWithTask, *v1.TriggerFromEventsResult, error)

func (p *SchedulingPool) SetTenants(tenants []*sqlcv1.Tenant)

type SlotCp ¶

type SlotCp = scheduling.SlotCp

type SlotUtilization ¶

type SlotUtilization = scheduling.SlotUtilization

type SnapshotInput ¶

type SnapshotInput = scheduling.SnapshotInput

type WorkerCp ¶

type WorkerCp = scheduling.WorkerCp

type WorkerLabelPair ¶

type WorkerLabelPair struct {
	Key   string
	Value string
}

type WorkerPromLabels ¶

type WorkerPromLabels struct {
	ID   uuid.UUID
	Name string
}