92 lines
4.2 KiB
Go
92 lines
4.2 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
|
|
}
|
|
|
|
const transientDispatchRetryDelay = time.Second
|
|
|
|
func retryClaimedDispatch(ctx context.Context, store ports.WorkerStore, dispatchID, workerID string, now time.Time) error {
|
|
return store.RetryDispatch(ctx, dispatchID, workerID, now.UnixMilli(), transientDispatchRetryDelay)
|
|
}
|
|
|
|
// RecoverOnlyWorkerCycle keeps durable recovery alive while the real executor
|
|
// capability is absent. It intentionally does not claim a dispatch: a missing
|
|
// executor must never be represented as successful work.
|
|
func RecoverOnlyWorkerCycle(ctx context.Context, store ports.WorkerStore, now time.Time, staleAfter time.Duration) (WorkerCycleResult, error) {
|
|
recovered, err := store.RecoverStaleClaims(ctx, now.UnixMilli(), staleAfter.Milliseconds())
|
|
return WorkerCycleResult{RecoveredClaims: recovered, Idle: err == nil}, err
|
|
}
|
|
|
|
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 {
|
|
_ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now)
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
|
|
}
|
|
if _, err := ExecuteDryRunDispatch(ctx, store, intent, workerID, token, now, leaseTTL); err != nil {
|
|
_ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now)
|
|
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
|
|
}
|
|
|
|
func ExecuteWorkerCycleWithExecutor(ctx context.Context, store ports.WorkerStore, executor ports.AgentExecutor, 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 {
|
|
_ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now)
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
|
|
}
|
|
result, err := ExecuteAgentDispatch(ctx, store, executor, intent, workerID, token, now, leaseTTL)
|
|
if err != nil {
|
|
_ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now)
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
|
|
}
|
|
if result.State != work.AgentExecutionCompleted {
|
|
if err := retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now); err != nil {
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
|
|
}
|
|
return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, nil
|
|
}
|
|
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
|
|
}
|