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

v1

package
v0.112.16 Latest Latest
Warning

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

 Go to latest Published: Oct 20, 2026 License: MIT Imports: 33 Imported by: 0

Documentation

Index

Constants

View Source 
const MAX_RATE_LIMIT_UPDATE_FREQUENCY = 500 * time.Millisecond // avoid boundary conditions on 1 second polls

Variables

View Source 
var ErrNoOptimisticSlots = scheduling.ErrNoOptimisticSlots
View Source 
var ErrTenantNotFound = scheduling.ErrTenantNotFound

aliased so errors.Is checks in the engine match either scheduler implementation

Functions

This section is empty.

Types

type AssignedItemWithTask added in v0.78.0 

type AssignedItemWithTask = scheduling.AssignedItemWithTask

type BatchScheduler added in v0.98.0 

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

func (*BatchScheduler) Cleanup added in v0.98.0 

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

func (*BatchScheduler) Start added in v0.98.0 

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

type ConcurrencyManager 

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

func (*ConcurrencyManager) Cleanup 

func (c *ConcurrencyManager) Cleanup()

func (*ConcurrencyManager) UpdateStrategy added in v0.105.20 

func (c *ConcurrencyManager) UpdateStrategy(next *sqlcv1.V1StepConcurrency) bool

UpdateStrategy applies a changed definition to the running manager in place, avoiding the cleanup/recreate cycle (topic re-acquisition and a full slot rehydration, plus the contention window where the old manager still holds its locks while the new one spins up). It returns false when the change is structural — the limit-strategy kind or the parent linkage changed, or the manager runs the SQL path (no in-memory index), which reads its definition from the construction-time strategy — in which case the caller falls back to a rebuild. Reconcile-goroutine only.

type ConcurrencyResults 

type ConcurrencyResults = scheduling.ConcurrencyResults

type Extensions 

type Extensions = scheduling.Extensions

type LabelPairPromLabels added in v0.94.8 

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 against the pool — so it is defined once in pkg/scheduling and aliased here.

type PrometheusExtension added in v0.72.2 

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

func NewPrometheusExtension added in v0.72.2 

func NewPrometheusExtension(promGate *prometheus.Gate) *PrometheusExtension

func (*PrometheusExtension) Cleanup added in v0.72.2 

func (p *PrometheusExtension) Cleanup() error

func (*PrometheusExtension) CleanupTenant added in v0.80.5 

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

func (*PrometheusExtension) PostAssign added in v0.72.2 

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

func (*PrometheusExtension) ReportSnapshot added in v0.72.2 

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

func (*PrometheusExtension) SetTenants added in v0.72.2 

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.

type Queuer 

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

func (*Queuer) Cleanup 

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 

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 (*SchedulingPool) AddExtension added in v0.100.18 

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

func (*SchedulingPool) GetConcurrencyResultsCh 

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

func (*SchedulingPool) GetResultsCh 

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

func (*SchedulingPool) NotifyConcurrency 

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

func (*SchedulingPool) NotifyNewConcurrencyStrategy added in v0.78.21 

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

func (*SchedulingPool) NotifyNewQueue added in v0.78.21 

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

func (*SchedulingPool) NotifyNewWorker added in v0.78.21 

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

func (*SchedulingPool) NotifyQueues 

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

func (*SchedulingPool) Replenish 

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

func (*SchedulingPool) RunOptimisticScheduling added in v0.78.0 

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 (*SchedulingPool) RunOptimisticSchedulingFromEvents added in v0.78.0 

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 (*SchedulingPool) SetTenants 

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 added in v0.94.8 

type WorkerLabelPair struct {
	Key   string
	Value string
}

type WorkerPromLabels added in v0.72.2 

type WorkerPromLabels struct {
	ID   uuid.UUID
	Name string
}

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.