Skip to content

feat: optimistic scheduling #2258

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
Open

feat: optimistic scheduling #2258

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
437ee21
feat: e2e working version of optimistic scheduling
abelanger5 Sep 7, 2025
cc1b59c
Merge remote-tracking branch 'origin/main' into belanger/optimistic-s…
abelanger5 Sep 7, 2025
8705e84
add migration
abelanger5 Sep 7, 2025
ba913d8
add optimistic to matrix
abelanger5 Sep 7, 2025
82d2b99
fix: ignore expirable goleak
abelanger5 Sep 7, 2025
b1a7aec
fix: case on err no rows when querying queue items
abelanger5 Sep 7, 2025
e4ce3d5
fix: trigger logic
abelanger5 Sep 7, 2025
fde8529
fix: proper casing on task-completed payload
abelanger5 Sep 8, 2025
d294219
fix: continue processing events after local scheduling
abelanger5 Sep 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 85 additions & 65 deletions cmd/hatchet-engine/engine/run.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,8 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
})
}

var localScheduler *schedulerv1.Scheduler

// FIXME: jobscontroller and workflowscontroller are deprecated service names, but there's not a clear upgrade
// path for old config files.
if sc.HasService("queue") || sc.HasService("jobscontroller") || sc.HasService("workflowscontroller") || sc.HasService("retention") || sc.HasService("ticker") {
Expand Down Expand Up @@ -272,6 +274,8 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
Name: "schedulerv1",
Fn: cleanup,
})

localScheduler = sv1
}

if sc.HasService("ticker") {
Expand Down Expand Up @@ -474,6 +478,8 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
ingestor.WithStepRunRepository(sc.EngineRepository.StepRun()),
ingestor.WithRepositoryV1(sc.V1),
ingestor.WithLogIngestionEnabled(sc.Runtime.LogIngestionEnabled),
ingestor.WithLocalScheduler(localScheduler),
ingestor.WithLocalDispatcher(d),
)

if err != nil {
Expand All @@ -486,6 +492,8 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
admin.WithMessageQueue(sc.MessageQueue),
admin.WithMessageQueueV1(sc.MessageQueueV1),
admin.WithEntitlementsRepository(sc.EntitlementRepository),
admin.WithLocalScheduler(localScheduler),
admin.WithLocalDispatcher(d),
)
if err != nil {
return nil, fmt.Errorf("could not create admin service: %w", err)
Expand All @@ -496,6 +504,8 @@ func runV0Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
adminv1.WithMessageQueue(sc.MessageQueueV1),
adminv1.WithEntitlementsRepository(sc.EntitlementRepository),
adminv1.WithAnalytics(sc.Analytics),
adminv1.WithLocalScheduler(localScheduler),
adminv1.WithLocalDispatcher(d),
)

if err != nil {
Expand Down Expand Up @@ -878,6 +888,75 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
})
}

var localScheduler *schedulerv1.Scheduler

if sc.HasService("all") || sc.HasService("scheduler") {
partitionCleanup, err := p.StartSchedulerPartition(ctx)

if err != nil {
return nil, fmt.Errorf("could not create create scheduler partition: %w", err)
}

teardown = append(teardown, Teardown{
Name: "scheduler partition",
Fn: partitionCleanup,
})

// create the dispatcher
s, err := scheduler.New(
scheduler.WithAlerter(sc.Alerter),
scheduler.WithMessageQueue(sc.MessageQueue),
scheduler.WithRepository(sc.EngineRepository),
scheduler.WithLogger(sc.Logger),
scheduler.WithPartition(p),
scheduler.WithQueueLoggerConfig(&sc.AdditionalLoggers.Queue),
scheduler.WithSchedulerPool(sc.SchedulingPool),
)

if err != nil {
return nil, fmt.Errorf("could not create dispatcher: %w", err)
}

cleanup, err := s.Start()

if err != nil {
return nil, fmt.Errorf("could not start dispatcher: %w", err)
}

teardown = append(teardown, Teardown{
Name: "scheduler",
Fn: cleanup,
})

sv1, err := schedulerv1.New(
schedulerv1.WithAlerter(sc.Alerter),
schedulerv1.WithMessageQueue(sc.MessageQueueV1),
schedulerv1.WithRepository(sc.EngineRepository),
schedulerv1.WithV2Repository(sc.V1),
schedulerv1.WithLogger(sc.Logger),
schedulerv1.WithPartition(p),
schedulerv1.WithQueueLoggerConfig(&sc.AdditionalLoggers.Queue),
schedulerv1.WithSchedulerPool(sc.SchedulingPoolV1),
)

if err != nil {
return nil, fmt.Errorf("could not create scheduler (v1): %w", err)
}

cleanup, err = sv1.Start()

if err != nil {
return nil, fmt.Errorf("could not start scheduler (v1): %w", err)
}

teardown = append(teardown, Teardown{
Name: "schedulerv1",
Fn: cleanup,
})

localScheduler = sv1
}
Comment on lines +891 to +958
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just moved this up and added the localScheduler assignment, but otherwise no code changes.


if sc.HasService("all") || sc.HasService("grpc-api") {
cacheInstance := cache.New(10 * time.Second)

Expand Down Expand Up @@ -931,6 +1010,8 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
ingestor.WithStepRunRepository(sc.EngineRepository.StepRun()),
ingestor.WithRepositoryV1(sc.V1),
ingestor.WithLogIngestionEnabled(sc.Runtime.LogIngestionEnabled),
ingestor.WithLocalScheduler(localScheduler),
ingestor.WithLocalDispatcher(d),
)

if err != nil {
Expand All @@ -943,6 +1024,8 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
admin.WithMessageQueue(sc.MessageQueue),
admin.WithMessageQueueV1(sc.MessageQueueV1),
admin.WithEntitlementsRepository(sc.EntitlementRepository),
admin.WithLocalScheduler(localScheduler),
admin.WithLocalDispatcher(d),
)

if err != nil {
Expand All @@ -954,6 +1037,8 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
adminv1.WithMessageQueue(sc.MessageQueueV1),
adminv1.WithEntitlementsRepository(sc.EntitlementRepository),
adminv1.WithAnalytics(sc.Analytics),
adminv1.WithLocalScheduler(localScheduler),
adminv1.WithLocalDispatcher(d),
)

if err != nil {
Expand Down Expand Up @@ -1025,71 +1110,6 @@ func runV1Config(ctx context.Context, sc *server.ServerConfig) ([]Teardown, erro
})
}

if sc.HasService("all") || sc.HasService("scheduler") {
partitionCleanup, err := p.StartSchedulerPartition(ctx)

if err != nil {
return nil, fmt.Errorf("could not create create scheduler partition: %w", err)
}

teardown = append(teardown, Teardown{
Name: "scheduler partition",
Fn: partitionCleanup,
})

// create the dispatcher
s, err := scheduler.New(
scheduler.WithAlerter(sc.Alerter),
scheduler.WithMessageQueue(sc.MessageQueue),
scheduler.WithRepository(sc.EngineRepository),
scheduler.WithLogger(sc.Logger),
scheduler.WithPartition(p),
scheduler.WithQueueLoggerConfig(&sc.AdditionalLoggers.Queue),
scheduler.WithSchedulerPool(sc.SchedulingPool),
)

if err != nil {
return nil, fmt.Errorf("could not create dispatcher: %w", err)
}

cleanup, err := s.Start()

if err != nil {
return nil, fmt.Errorf("could not start dispatcher: %w", err)
}

teardown = append(teardown, Teardown{
Name: "scheduler",
Fn: cleanup,
})

sv1, err := schedulerv1.New(
schedulerv1.WithAlerter(sc.Alerter),
schedulerv1.WithMessageQueue(sc.MessageQueueV1),
schedulerv1.WithRepository(sc.EngineRepository),
schedulerv1.WithV2Repository(sc.V1),
schedulerv1.WithLogger(sc.Logger),
schedulerv1.WithPartition(p),
schedulerv1.WithQueueLoggerConfig(&sc.AdditionalLoggers.Queue),
schedulerv1.WithSchedulerPool(sc.SchedulingPoolV1),
)

if err != nil {
return nil, fmt.Errorf("could not create scheduler (v1): %w", err)
}

cleanup, err = sv1.Start()

if err != nil {
return nil, fmt.Errorf("could not start scheduler (v1): %w", err)
}

teardown = append(teardown, Teardown{
Name: "schedulerv1",
Fn: cleanup,
})
}

teardown = append(teardown, Teardown{
Name: "telemetry",
Fn: func() error {
Expand Down
Loading
Loading