217 lines
8.3 KiB
Go
217 lines
8.3 KiB
Go
package store
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.ipao.vip/rogee/go-sip/internal/contract"
|
|
"git.ipao.vip/rogee/go-sip/internal/mq"
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
type taskControlPayload struct {
|
|
TaskID string `json:"task_id"`
|
|
Action string `json:"action"`
|
|
ExpectedRevision int64 `json:"expected_task_revision"`
|
|
Policy string `json:"active_call_policy"`
|
|
}
|
|
|
|
type controlTarget struct {
|
|
executionID, status string
|
|
revision int64
|
|
active bool
|
|
}
|
|
|
|
// HandleTaskControl persists the admission barrier, required remote work and
|
|
// original receipt atomically. Active calls remain accepted, never falsely
|
|
// reported applied before their Agent acknowledges the control.
|
|
func (s *Store) HandleTaskControl(raw []byte, routingKey string) (string, bool, error) {
|
|
command, err := contract.DecodeMQCommand(raw)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
if command.CommandType != "task.control" {
|
|
return "", false, errors.New("expected task.control")
|
|
}
|
|
route, err := tenant.NewDispatcherRoute(command.DispatcherID, command.TenantKey)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
if routingKey != route.InboundKey {
|
|
return "", false, ErrMessageScope
|
|
}
|
|
var payload taskControlPayload
|
|
if err := json.Unmarshal(command.Payload, &payload); err != nil {
|
|
return "", false, err
|
|
}
|
|
digest := sha256.Sum256(raw)
|
|
bodyHash := hex.EncodeToString(digest[:])
|
|
now := s.now().UTC()
|
|
stamp := now.Format(time.RFC3339Nano)
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
defer tx.Rollback()
|
|
if err := bindMQScope(tx, command.DispatcherID, command.TenantID, command.TenantKey); err != nil {
|
|
return "", false, err
|
|
}
|
|
var oldHash, responseID string
|
|
err = tx.QueryRow(`SELECT i.body_hash,r.response_id FROM inbox i JOIN mq_command_receipts r ON r.tenant_id=i.tenant_id AND r.command_id=i.command_id WHERE i.tenant_id=? AND i.command_id=?`, command.TenantID, command.CommandID).Scan(&oldHash, &responseID)
|
|
if err == nil {
|
|
if oldHash != bodyHash {
|
|
return "", false, ErrIdempotencyConflict
|
|
}
|
|
if _, err := tx.Exec(`UPDATE outbox SET status='pending',published_at=NULL WHERE event_id=? AND status='published'`, responseID); err != nil {
|
|
return "", false, err
|
|
}
|
|
return responseID, true, tx.Commit()
|
|
}
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return "", false, err
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO inbox(command_id,tenant_id,tenant_key,command_type,body_hash,body,status,received_at,persisted_at) VALUES(?,?,?,?,?,?,'persisted',?,?)`, command.CommandID, command.TenantID, command.TenantKey, command.CommandType, bodyHash, raw, stamp, stamp); err != nil {
|
|
return "", false, err
|
|
}
|
|
status, reason := "applied", "applied"
|
|
expired, err := contract.NotAfterExpired(command.NotAfter, now)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
var targets []controlTarget
|
|
if expired {
|
|
status, reason = "rejected", "expired"
|
|
} else {
|
|
rows, err := tx.Query(`SELECT execution_id,status,task_revision,
|
|
(status IN ('running','unknown')
|
|
OR EXISTS(SELECT 1 FROM reservations r WHERE r.execution_id=tasks.execution_id AND r.state='unknown')
|
|
OR (EXISTS(SELECT 1 FROM execution_agents a WHERE a.execution_id=tasks.execution_id)
|
|
AND (status='reserved' OR EXISTS(SELECT 1 FROM reservations r WHERE r.execution_id=tasks.execution_id AND r.state='held'))))
|
|
FROM tasks WHERE tenant_id=? AND tenant_key=? AND task_id=?`, command.TenantID, command.TenantKey, payload.TaskID)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
for rows.Next() {
|
|
var target controlTarget
|
|
if err := rows.Scan(&target.executionID, &target.status, &target.revision, &target.active); err != nil {
|
|
rows.Close()
|
|
return "", false, err
|
|
}
|
|
targets = append(targets, target)
|
|
}
|
|
rowErr := rows.Err()
|
|
closeErr := rows.Close()
|
|
if err := errors.Join(rowErr, closeErr); err != nil {
|
|
return "", false, err
|
|
}
|
|
if len(targets) == 0 {
|
|
status, reason = "rejected", "not_found"
|
|
}
|
|
var pending int
|
|
if err := tx.QueryRow(`SELECT COUNT(*) FROM mq_task_controls WHERE tenant_id=? AND task_id=? AND state='pending'`, command.TenantID, payload.TaskID).Scan(&pending); err != nil {
|
|
return "", false, err
|
|
}
|
|
if pending != 0 {
|
|
status, reason = "rejected", "control_pending"
|
|
}
|
|
for _, target := range targets {
|
|
if target.revision != payload.ExpectedRevision {
|
|
status, reason = "rejected", "revision_conflict"
|
|
break
|
|
}
|
|
if (target.status == "stopped" && payload.Action != "stop") || (payload.Action == "resume" && target.status != "paused") {
|
|
status, reason = "rejected", "invalid_state"
|
|
break
|
|
}
|
|
}
|
|
}
|
|
revision := payload.ExpectedRevision
|
|
if status != "rejected" {
|
|
hasActive := false
|
|
for _, target := range targets {
|
|
hasActive = hasActive || target.active
|
|
}
|
|
state := "applied"
|
|
if hasActive {
|
|
state = "pending"
|
|
status, reason = "accepted", "accepted"
|
|
} else {
|
|
revision++
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO mq_task_controls(tenant_id,command_id,tenant_key,task_id,expected_revision,action,active_call_policy,state) VALUES(?,?,?,?,?,?,?,?)`, command.TenantID, command.CommandID, command.TenantKey, payload.TaskID, payload.ExpectedRevision, payload.Action, payload.Policy, state); err != nil {
|
|
return "", false, err
|
|
}
|
|
for _, target := range targets {
|
|
next := target.status
|
|
switch payload.Action {
|
|
case "pause":
|
|
next = "paused"
|
|
case "resume":
|
|
next = "accepted"
|
|
case "stop":
|
|
next = "stopped"
|
|
default:
|
|
return "", false, errors.New("unsupported task control action")
|
|
}
|
|
if target.status == "finished" {
|
|
next = target.status
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO mq_task_control_targets(tenant_id,command_id,execution_id,original_status,applied) VALUES(?,?,?,?,?)`, command.TenantID, command.CommandID, target.executionID, target.status, !target.active); err != nil {
|
|
return "", false, err
|
|
}
|
|
if _, err := tx.Exec(`UPDATE tasks SET status=?,task_revision=?,updated_at=? WHERE execution_id=? AND tenant_id=? AND task_revision=?`, next, revision, stamp, target.executionID, command.TenantID, payload.ExpectedRevision); err != nil {
|
|
return "", false, err
|
|
}
|
|
if !target.active && target.status == "reserved" {
|
|
// No Agent assignment exists, so no remote execution was submitted.
|
|
// Release these local reservations in the same control transaction.
|
|
rows, err := tx.Query(`SELECT reservation_id FROM reservations WHERE execution_id=? AND state='held'`, target.executionID)
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
var reservations []string
|
|
for rows.Next() {
|
|
var id string
|
|
if err := rows.Scan(&id); err != nil {
|
|
rows.Close()
|
|
return "", false, err
|
|
}
|
|
reservations = append(reservations, id)
|
|
}
|
|
if err := errors.Join(rows.Err(), rows.Close()); err != nil {
|
|
return "", false, err
|
|
}
|
|
for _, id := range reservations {
|
|
if err := s.finalizeReservationTx(tx, id, target.executionID, false); err != nil {
|
|
return "", false, err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
responseID = uuid.NewSHA1(uuid.NameSpaceURL, []byte("task.control:"+command.DispatcherID+":"+command.TenantID+":"+command.CommandID)).String()
|
|
result := map[string]any{"command_id": command.CommandID, "command_type": command.CommandType, "status": status, "reason_code": reason, "task_id": payload.TaskID, "requested_task_revision": payload.ExpectedRevision}
|
|
if status == "applied" {
|
|
result["applied_task_revision"] = revision
|
|
}
|
|
body, err := (contract.EventBuilder{TenantID: command.TenantID, TenantKey: command.TenantKey, TraceID: command.TraceID, EventType: "command.result", Aggregate: "command", AggregateID: command.CommandID, Version: 1, Payload: result}).MarshalMQ(command.DispatcherID, now, responseID)
|
|
if err != nil {
|
|
return "", false, fmt.Errorf("control receipt: %w", err)
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO outbox(event_id,tenant_key,exchange,routing_key,body,status,created_at) VALUES(?,?,?,?,?,'pending',?)`, responseID, command.TenantKey, mq.EventExchange, route.OutboundKey, body, stamp); err != nil {
|
|
return "", false, err
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO mq_command_receipts(tenant_id,command_id,response_id) VALUES(?,?,?)`, command.TenantID, command.CommandID, responseID); err != nil {
|
|
return "", false, err
|
|
}
|
|
return responseID, false, tx.Commit()
|
|
}
|