k2safe/orbit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: make runtime scanning tunable within safe bounds

Browse Source
This commit is contained in:
k2safe
2026-09-20 18:56:09 +08:00
parent 6d194122af
commit df6773be47
8 changed files with 94 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
cmd/orbit/main.go
View File

@@ -108,7 +108,7 @@ func run() error {
workflowDone := make(chan struct{})
go func() {
defer close(workflowDone)
ticker := time.NewTicker(time.Second)
ticker := time.NewTicker(cfg.WorkflowTickInterval)
defer ticker.Stop()
for {
select {
@@ -117,7 +117,7 @@ func run() error {
case <-ticker.C:
started := time.Now()
call, cancel := context.WithTimeout(workflowCtx, 5*time.Second)
_, err := workflows.Tick(call, 100)
_, err := workflows.Tick(call, cfg.RuntimeBatchLimit)
cancel()
if metrics != nil {
metrics.RecordBatch("workflow", started, err)
@@ -133,7 +133,7 @@ func run() error {
scheduleDone := make(chan struct{})
go func() {
defer close(scheduleDone)
ticker := time.NewTicker(time.Second)
ticker := time.NewTicker(cfg.ScheduleTickInterval)
defer ticker.Stop()
for {
select {
@@ -142,7 +142,7 @@ func run() error {
case <-ticker.C:
started := time.Now()
call, cancel := context.WithTimeout(scheduleCtx, 5*time.Second)
_, err := schedules.Tick(call, 100)
_, err := schedules.Tick(call, cfg.RuntimeBatchLimit)
cancel()
if metrics != nil {
metrics.RecordBatch("schedule", started, err)
@@ -163,7 +163,7 @@ func run() error {
recoveryCtx, stopRecovery := context.WithCancel(root)
go func() {
defer close(recoveryDone)
ticker := time.NewTicker(2 * time.Second)
ticker := time.NewTicker(cfg.RecoveryTickInterval)
defer ticker.Stop()
for {
select {
@@ -172,7 +172,7 @@ func run() error {
case <-ticker.C:
started := time.Now()
call, cancel := context.WithTimeout(recoveryCtx, 5*time.Second)
_, err := executions.Recover(call, 100)
_, err := executions.Recover(call, cfg.RuntimeBatchLimit)
cancel()
if metrics != nil {
metrics.RecordBatch("execution_recovery", started, err)
@@ -243,7 +243,7 @@ func run() error {
slog.Info("Orbit metrics server starting", "metrics_address", metricsListener.Addr().String())
}
go func() { done <- server.Serve(listener) }()
slog.Info("Orbit HTTP server starting", "address", listener.Addr().String(), "scope", "health, task, queue, schedule, subscription, workflow and Worker APIs", "node_id", node)
slog.Info("Orbit HTTP server starting", "address", listener.Addr().String(), "scope", "health, task, queue, schedule, subscription, workflow and Worker APIs", "node_id", node, "workflow_tick_interval", cfg.WorkflowTickInterval.String(), "schedule_tick_interval", cfg.ScheduleTickInterval.String(), "recovery_tick_interval", cfg.RecoveryTickInterval.String(), "runtime_batch_limit", cfg.RuntimeBatchLimit)
select {
case err := <-done:
if errors.Is(err, http.ErrServerClosed) {