44 lines
1.7 KiB
Go
44 lines
1.7 KiB
Go
package application
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/contracts/work"
|
|
"git.cin.su/m/Assistant/modules/work/ports"
|
|
)
|
|
|
|
type WorkerCycleResult struct {
|
|
RecoveredClaims int64
|
|
DispatchID string
|
|
Completed bool
|
|
Idle bool
|
|
}
|
|
|
|
func ExecuteWorkerCycle(ctx context.Context, store ports.WorkerStore, workerID string, token work.AgentFencingToken, now time.Time, staleAfter, leaseTTL time.Duration) (WorkerCycleResult, error) {
|
|
recovered, err := store.RecoverStaleClaims(ctx, now.UnixMilli(), staleAfter.Milliseconds())
|
|
if err != nil {
|
|
return WorkerCycleResult{}, err
|
|
}
|
|
item, err := store.ClaimNextDispatch(ctx, workerID, now.UnixMilli())
|
|
if err != nil {
|
|
return WorkerCycleResult{RecoveredClaims: recovered}, err
|
|
}
|
|
if item.DispatchID == "" {
|
|
return WorkerCycleResult{RecoveredClaims: recovered, Idle: true}, nil
|
|
}
|
|
intent, err := store.LoadDispatchIntent(ctx, item.DispatchID)
|
|
if err != nil {
|
|
_ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli())
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
|
|
}
|
|
if _, err := ExecuteDryRunDispatch(ctx, store, intent, workerID, token, now, leaseTTL); err != nil {
|
|
_ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli())
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
|
|
}
|
|
if err := store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchCompleted, now.UnixMilli()); err != nil {
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
|
|
}
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID, Completed: true}, nil
|
|
}
|