114 lines
3.8 KiB
Go
114 lines
3.8 KiB
Go
package application
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/contracts/work"
|
|
"git.cin.su/m/Assistant/modules/work/domain"
|
|
"git.cin.su/m/Assistant/modules/work/ports"
|
|
)
|
|
|
|
func ExecuteAgentDispatch(ctx context.Context, store ports.ExecutionStore, executor ports.AgentExecutor, intent work.AgentDispatchIntent, workerID string, token work.AgentFencingToken, now time.Time, ttl time.Duration) (work.AgentExecutionResult, error) {
|
|
if executor == nil {
|
|
return work.AgentExecutionResult{}, ports.ErrAgentExecutorUnavailable
|
|
}
|
|
if err := domain.ValidateDispatchIntent(intent); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
lease, err := domain.AcquireAgentLease(intent.RunID, workerID, token, now, ttl)
|
|
if err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.SaveDispatchIntent(ctx, intent); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.SaveAgentLease(ctx, lease); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
heartbeat := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 1, Kind: work.AgentCallbackHeartbeat, ObservedAt: now}
|
|
if err := domain.ValidateAgentCallback(heartbeat, lease, 0); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.AcceptAgentCallback(ctx, heartbeat); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
executionCtx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
var stateMu sync.Mutex
|
|
currentLease := lease
|
|
lastSequence := heartbeat.Sequence
|
|
heartbeatDone := make(chan struct{})
|
|
heartbeatStopped := make(chan struct{})
|
|
heartbeatFailure := make(chan error, 1)
|
|
interval := ttl / 3
|
|
if interval <= 0 {
|
|
return work.AgentExecutionResult{}, fmt.Errorf("invalid heartbeat interval")
|
|
}
|
|
go func() {
|
|
defer close(heartbeatStopped)
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-heartbeatDone:
|
|
return
|
|
case observedAt := <-ticker.C:
|
|
stateMu.Lock()
|
|
nextLease, renewErr := domain.RenewAgentLease(currentLease, workerID, token, observedAt, ttl)
|
|
if renewErr == nil {
|
|
renewErr = store.RenewAgentLease(executionCtx, nextLease, nextLease.ExpiresAt)
|
|
}
|
|
if renewErr == nil {
|
|
lastSequence++
|
|
heartbeat = work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: lastSequence, Kind: work.AgentCallbackHeartbeat, ObservedAt: observedAt}
|
|
renewErr = domain.ValidateAgentCallback(heartbeat, nextLease, lastSequence-1)
|
|
}
|
|
if renewErr == nil {
|
|
renewErr = store.AcceptAgentCallback(executionCtx, heartbeat)
|
|
}
|
|
if renewErr == nil {
|
|
currentLease = nextLease
|
|
}
|
|
stateMu.Unlock()
|
|
if renewErr != nil {
|
|
select {
|
|
case heartbeatFailure <- renewErr:
|
|
default:
|
|
}
|
|
cancel()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
result, err := executor.Execute(executionCtx, work.AgentExecutionRequest{Intent: intent, WorkerID: workerID, FencingToken: token, LeaseExpiresAt: lease.ExpiresAt})
|
|
close(heartbeatDone)
|
|
<-heartbeatStopped
|
|
select {
|
|
case heartbeatErr := <-heartbeatFailure:
|
|
return work.AgentExecutionResult{}, heartbeatErr
|
|
default:
|
|
}
|
|
if err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
stateMu.Lock()
|
|
lease = currentLease
|
|
stateMu.Unlock()
|
|
observedAt := time.UnixMilli(result.ObservedAtUnixMillis)
|
|
callback := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: lastSequence + 1, Kind: work.AgentCallbackResult, ObservedAt: observedAt}
|
|
if result.State == work.AgentExecutionWaiting {
|
|
callback.Kind = work.AgentCallbackCheckpoint
|
|
}
|
|
if err := domain.ValidateAgentCallback(callback, lease, lastSequence); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if err := store.AcceptAgentCallback(ctx, callback); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
return result, nil
|
|
}
|