diff --git a/contracts/control_receipt_test.go b/contracts/control_receipt_test.go new file mode 100644 index 0000000..e7c3bb3 --- /dev/null +++ b/contracts/control_receipt_test.go @@ -0,0 +1,52 @@ +package contracts + +import ( + "encoding/json" + "testing" +) + +func TestTaskControlReceiptFields(t *testing.T) { + schema, err := CompileCurrent("mq-task-control-receipt.schema.json") + if err != nil { + t.Fatal(err) + } + envelope := func(payload map[string]any) map[string]any { + return map[string]any{"event_id": "control-fields", "event_type": "task.control", "dispatcher_id": "c046b893-8628-4589-ae50-619d049248a6", "tenant_id": 1001, "payload": payload} + } + applied := map[string]any{"task_id": "task-a", "action": "pause", "status": "applied", "reason_code": "applied", "task_state": "paused"} + if err := schema.Validate(envelope(applied)); err != nil { + t.Fatalf("requested success structure rejected: %v", err) + } + for key := range applied { + raw, _ := json.Marshal(applied) + var missing map[string]any + _ = json.Unmarshal(raw, &missing) + delete(missing, key) + if err := schema.Validate(envelope(missing)); err == nil { + t.Errorf("success accepted without %s", key) + } + } + for _, payload := range []map[string]any{ + {"status": "rejected", "reason_code": "invalid_data", "reason_message": "issued_at missing"}, + {"task_id": "task-a", "action": "pause", "status": "rejected", "reason_code": "control_rejected", "task_state": "stopped", "reason_message": "task stopped"}, + {"task_id": "task-a", "status": "rejected", "reason_code": "invalid_data", "task_state": "running", "reason_message": "invalid action"}, + } { + if err := schema.Validate(envelope(payload)); err != nil { + t.Errorf("rejection with independently known fields rejected: %v", err) + } + } + for _, payload := range []map[string]any{ + {"status": "applied"}, + {"task_id": "task-a", "action": "pause", "status": "applied", "reason_code": "invalid_data", "task_state": "paused"}, + {"status": "rejected", "reason_code": "invalid_data", "reason_message": ""}, + {"status": "rejected", "reason_message": "invalid data"}, + {"task_id": nil, "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"}, + {"task_id": "bad task", "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"}, + {"action": "unknown", "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"}, + {"task_state": "invented", "status": "rejected", "reason_code": "invalid_data", "reason_message": "invalid data"}, + } { + if err := schema.Validate(envelope(payload)); err == nil { + t.Errorf("invalid payload accepted: %+v", payload) + } + } +} diff --git a/contracts/schema b/contracts/schema index 8e6c332..e750fe4 160000 --- a/contracts/schema +++ b/contracts/schema @@ -1 +1 @@ -Subproject commit 8e6c332bb9d5b3d331cef3e5cbc35a1b8b4b7c86 +Subproject commit e750fe4618e6472821ceaa74f6c6227d11f998e3 diff --git a/docs/contracts.md b/docs/contracts.md index be5196c..536a09a 100644 --- a/docs/contracts.md +++ b/docs/contracts.md @@ -7,6 +7,7 @@ SaaS 开发人员与本项目共同维护业务规范、Schema、MQ 拓扑和正 本项目的 `contracts/schema` 是 Git submodule，只记录消费的确定提交，不维护另一份本地定义。 原 `contracts/local` 已移除；`docs/thirds/saas-dispatcher.md` 只提供导航。 内部 `proto/agent/agent.proto` 仍属于本项目，不移入 SaaS 业务契约。 +任务控制回执及逐字段拒绝反馈规则统一见共享仓库的 `mq-task-control-receipt.schema.json`、`saas-dispatcher.md` 和 `examples/mq-control-{ack,rejected}.json`；SaaS 须同步采用这些入口，不以历史归档提案或旧的仅 status 回执作为现行结构。 | 位置 | 用途 | | --- | --- | diff --git a/internal/contract/rejection_test.go b/internal/contract/rejection_test.go index 4d8644a..208d062 100644 --- a/internal/contract/rejection_test.go +++ b/internal/contract/rejection_test.go @@ -16,6 +16,9 @@ func TestRejectedFeedbackRequiresReasonAndPreservesOriginalType(t *testing.T) { if kind == "call.execute" { body["payload"].(map[string]any)["reason_code"] = nil } + if kind == "task.control" { + body["payload"].(map[string]any)["reason_code"] = "invalid_data" + } raw, err := json.Marshal(body) if err != nil { t.Fatal(err) diff --git a/internal/dispatcher/control.go b/internal/dispatcher/control.go index f4f1d2e..87fde78 100644 --- a/internal/dispatcher/control.go +++ b/internal/dispatcher/control.go @@ -60,6 +60,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re TaskID string `json:"task_id"` Action string `json:"action"` Reason string `json:"reason"` + Status string `json:"status"` Options struct { ActiveCallPolicy string `json:"active_call_policy"` } `json:"options"` @@ -68,7 +69,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re if err := json.Unmarshal(body, &event); err != nil { return fmt.Errorf("decode task.control: %w", err) } - if event.EventType != "task.control" || event.DispatcherID != c.DispatcherID || event.Payload.TaskID == "" || event.Payload.Action == "" { + if event.EventType != "task.control" || event.DispatcherID != c.DispatcherID || event.Payload.TaskID == "" || event.Payload.Action == "" || event.Payload.Status != "" { return errors.New("control event type, task, or Dispatcher owner mismatch") } defer func() { @@ -134,7 +135,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re if event.Payload.Action == "start" { if err := c.Store.CompleteStart(snapshot, event.EventID); err != nil { if errors.Is(err, store.ErrControlRejected) { - return c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, err.Error()) + return c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, event.Payload.TaskID, event.Payload.Action, err.Error()) } return fmt.Errorf("start task %q: %w", event.Payload.TaskID, err) } @@ -153,7 +154,7 @@ func (c *ControlController) ProcessControl(ctx context.Context, body []byte) (re } if err := c.Store.PrepareControl(event.DispatcherID, event.TenantID, event.Payload.TaskID, event.Payload.Action); err != nil { if errors.Is(err, store.ErrControlRejected) { - if rejectErr := c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, err.Error()); rejectErr != nil { + if rejectErr := c.Store.RejectControl(event.DispatcherID, event.TenantID, event.EventID, event.Payload.TaskID, event.Payload.Action, err.Error()); rejectErr != nil { return fmt.Errorf("persist rejected control %q: %w", event.EventID, rejectErr) } return nil diff --git a/internal/mq/broker.go b/internal/mq/broker.go index 53bf7c3..17b36bd 100644 --- a/internal/mq/broker.go +++ b/internal/mq/broker.go @@ -415,6 +415,7 @@ func (b *Broker) validateInbound(queue, routingKey string, body []byte) error { TaskID string `json:"task_id"` Action string `json:"action"` Callee string `json:"callee"` + Status string `json:"status"` TrunkID string `json:"trunk_id"` Revision int64 `json:"revision"` } `json:"payload"` @@ -422,6 +423,9 @@ func (b *Broker) validateInbound(queue, routingKey string, body []byte) error { if err := json.Unmarshal(body, &identity); err != nil { return fmt.Errorf("decode MQ message identity: %w", err) } + if identity.Payload.Status != "" { + return errors.New("outbound receipt cannot enter an inbound command queue") + } if identity.DispatcherID != b.dispatcherID { return errors.New("MQ message belongs to another Dispatcher") } diff --git a/internal/mq/broker_integration_test.go b/internal/mq/broker_integration_test.go index 9d326fb..47488a7 100644 --- a/internal/mq/broker_integration_test.go +++ b/internal/mq/broker_integration_test.go @@ -3,6 +3,7 @@ package mq import ( + "bytes" "context" "encoding/json" "errors" @@ -304,16 +305,69 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) { var receipt struct { EventID string `json:"event_id"` Payload struct { - Status string `json:"status"` - Reason string `json:"reason_message"` + TaskID string `json:"task_id"` + Action string `json:"action"` + Status string `json:"status"` + ReasonCode string `json:"reason_code"` + TaskState string `json:"task_state"` + Reason string `json:"reason_message"` } `json:"payload"` } - if err := json.Unmarshal(rejected.Body, &receipt); err != nil || receipt.EventID != "original-missing-issued-at" || receipt.Payload.Status != "rejected" || !strings.Contains(receipt.Payload.Reason, "issued_at") || strings.Contains(string(rejected.Body), "PRIVATE ORIGINAL") { + if err := json.Unmarshal(rejected.Body, &receipt); err != nil || receipt.EventID != "original-missing-issued-at" || receipt.Payload.TaskID != "task-asr" || receipt.Payload.Action != "resume" || receipt.Payload.TaskState != "running" || receipt.Payload.ReasonCode != "invalid_data" || receipt.Payload.Status != "rejected" || !strings.Contains(receipt.Payload.Reason, "issued_at") || strings.Contains(string(rejected.Body), "PRIVATE ORIGINAL") { t.Fatalf("untruthful/unsafe rejected feedback: %s %v", rejected.Body, err) } if err := rejected.Ack(false); err != nil { t.Fatal(err) } + // Inspect the actual queued success body as well as the correlated + // structural rejection above. This is isolated MQ, not external SaaS. + if err := s.PrepareControl(ids[0], 1001, "task-asr", "pause"); err != nil { + t.Fatal(err) + } + if err := s.CompleteControl(ids[0], 1001, "task-asr", "pause", "control-applied-fields"); err != nil { + t.Fatal(err) + } + feedback, err = s.ListPendingOutbox(ids[0]) + if err != nil { + t.Fatal(err) + } + var appliedBody []byte + for _, row := range feedback { + if row.EventID == "control-applied-fields" { + appliedBody = row.Body + } + } + if len(appliedBody) == 0 { + t.Fatal("applied receipt missing") + } + if err := brokers[0].Publish(context.Background(), resultDestination.Exchange, resultDestination.BindingKey, appliedBody); err != nil { + t.Fatal(err) + } + applied, ok, err := admin.Get(shared, false) + if err != nil || !ok || !bytes.Equal(applied.Body, appliedBody) { + t.Fatalf("success receipt changed or missing on MQ: %t %v", ok, err) + } + var success struct { + Payload map[string]any `json:"payload"` + } + if err := json.Unmarshal(applied.Body, &success); err != nil { + t.Fatal(err) + } + want := map[string]any{"task_id": "task-asr", "action": "pause", "status": "applied", "reason_code": "applied", "task_state": "paused"} + if len(success.Payload) != len(want) { + t.Fatalf("success receipt must contain exactly five fields: %+v", success.Payload) + } + for k, v := range want { + if success.Payload[k] != v { + t.Errorf("MQ success %s: got %v, want %v", k, success.Payload[k], v) + } + } + if err := applied.Ack(false); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(ids[0], "control-applied-fields"); err != nil { + t.Fatal(err) + } for _, id := range ids { route, _ := tenant.ResultRoute(id) if err := admin.QueueUnbind(shared, route.BindingKey, route.Exchange, nil); err != nil { diff --git a/internal/mq/inbound_test.go b/internal/mq/inbound_test.go index f394b37..ae15a9c 100644 --- a/internal/mq/inbound_test.go +++ b/internal/mq/inbound_test.go @@ -22,6 +22,8 @@ func TestBrokerInboundRequiresExactQueueRouteOwnerAndSchema(t *testing.T) { wantError bool }{ {"control", control.Queue, control.BindingKey, "mq-control", false}, + {"success receipt on control", control.Queue, control.BindingKey, "mq-control-ack", true}, + {"rejected receipt on control", control.Queue, control.BindingKey, "mq-control-rejected", true}, {"sip change", control.Queue, control.BindingKey, "mq-sip-change", false}, {"execute", task.Queue, task.BindingKey, "mq-execute", false}, {"wrong queue", "agent-call.d.c046b893-8628-4589-ae50-619d049248a7.task.task-asr.v1", task.BindingKey, "mq-execute", true}, diff --git a/internal/store/control_ack_test.go b/internal/store/control_ack_test.go index 99cb628..567bdeb 100644 --- a/internal/store/control_ack_test.go +++ b/internal/store/control_ack_test.go @@ -88,7 +88,7 @@ func TestStopRejectsResumeAndSuppressesOldCommands(t *testing.T) { if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "resume"); err == nil { t.Fatal("stopped task accepted resume") } - if err := s.RejectControl(currentDispatcherID, 1001, "resume-1", "task is stopped"); err != nil { + if err := s.RejectControl(currentDispatcherID, 1001, "resume-1", "task-asr", "resume", "task is stopped"); err != nil { t.Fatal(err) } outbox, err := s.ListPendingOutbox(currentDispatcherID) diff --git a/internal/store/control_flow.go b/internal/store/control_flow.go index 74eff00..f2cabf2 100644 --- a/internal/store/control_flow.go +++ b/internal/store/control_flow.go @@ -38,7 +38,7 @@ func (s *Store) CompleteStart(snapshot configread.Snapshot, eventID string) erro if err := applyDiscoveredTasks(tx, task.DispatcherID, []configread.DiscoveredTask{{TenantID: task.TenantID, TaskID: task.TaskID, TaskRevision: task.TaskRevision, Status: task.Status}}); err != nil { return err } - if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, "applied"); err != nil { + if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, task.TaskID, "start", "applied", ""); err != nil { return err } if err := tx.Commit(); err != nil { @@ -64,7 +64,7 @@ func (s *Store) CompleteEdit(snapshot configread.Snapshot, eventID string) (stri return "", fmt.Errorf("read edit task state: %w", err) } if errors.Is(err, sql.ErrNoRows) || present != 1 || state == "stopped" || state == "stopping" || status == "stopped" || task.Status == "stopped" { - if err := enqueueControlAckReason(tx, task.DispatcherID, task.TenantID, eventID, "rejected", "task is absent or stopped; edits are not allowed"); err != nil { + if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, task.TaskID, "edit", "rejected", "task is absent or stopped; edits are not allowed"); err != nil { return "", err } if err := tx.Commit(); err != nil { @@ -78,7 +78,7 @@ func (s *Store) CompleteEdit(snapshot configread.Snapshot, eventID string) (stri if _, err := tx.Exec(`UPDATE dispatcher_tasks SET task_revision=? WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, task.TaskRevision, task.DispatcherID, task.TenantID, task.TaskID); err != nil { return "", fmt.Errorf("persist edited task revision: %w", err) } - if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, "applied"); err != nil { + if err := enqueueControlAck(tx, task.DispatcherID, task.TenantID, eventID, task.TaskID, "edit", "applied", ""); err != nil { return "", err } if err := tx.Commit(); err != nil { @@ -181,7 +181,7 @@ func (s *Store) CompleteControl(dispatcherID string, tenantID int64, taskID, act if _, err := tx.Exec(`UPDATE dispatcher_tasks SET control_state=? WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, final, dispatcherID, tenantID, taskID); err != nil { return fmt.Errorf("persist applied task control: %w", err) } - if err := enqueueControlAck(tx, dispatcherID, tenantID, eventID, "applied"); err != nil { + if err := enqueueControlAck(tx, dispatcherID, tenantID, eventID, taskID, action, "applied", ""); err != nil { return err } if err := tx.Commit(); err != nil { @@ -190,13 +190,13 @@ func (s *Store) CompleteControl(dispatcherID string, tenantID int64, taskID, act return nil } -func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID, reason string) error { +func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID, taskID, action, reason string) error { tx, err := s.db.Begin() if err != nil { return err } defer tx.Rollback() - if err := enqueueControlAckReason(tx, dispatcherID, tenantID, eventID, "rejected", reason); err != nil { + if err := enqueueControlAck(tx, dispatcherID, tenantID, eventID, taskID, action, "rejected", reason); err != nil { return err } if err := tx.Commit(); err != nil { @@ -205,11 +205,7 @@ func (s *Store) RejectControl(dispatcherID string, tenantID int64, eventID, reas return nil } -func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, status string) error { - return enqueueControlAckReason(tx, dispatcherID, tenantID, eventID, status, "") -} - -func enqueueControlAckReason(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, status, reason string) error { +func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID, taskID, action, status, reason string) error { if status == "rejected" && strings.TrimSpace(reason) == "" { return errors.New("rejected task control requires a concrete reason") } @@ -220,37 +216,53 @@ func enqueueControlAckReason(tx *sql.Tx, dispatcherID string, tenantID int64, ev if err != nil { return err } + state, err := controlReceiptState(tx, dispatcherID, tenantID, taskID) + if err != nil { + return err + } + if status == "applied" && state == "" { + return errors.New("applied task control receipt requires a durable task state") + } + reasonCode := "applied" + if status == "rejected" { + reasonCode = "control_rejected" + } + payload := controlReceiptPayload{TaskID: taskID, Action: action, Status: status, ReasonCode: reasonCode, TaskState: state, ReasonMessage: reason} + // A retransmission reports the original state, not a later task state. + var oldType string + var oldBody []byte + err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&oldType, &oldBody) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("inspect existing task control outbox: %w", err) + } + if err == nil { + var previous struct { + Payload controlReceiptPayload `json:"payload"` + } + if err := json.Unmarshal(oldBody, &previous); err != nil { + return fmt.Errorf("decode original control receipt: %w", err) + } + payload.TaskState = previous.Payload.TaskState + } body, err := json.Marshal(struct { - EventID string `json:"event_id"` - EventType string `json:"event_type"` - DispatcherID string `json:"dispatcher_id"` - TenantID int64 `json:"tenant_id"` - Payload struct { - Status string `json:"status"` - Reason string `json:"reason_message,omitempty"` - } `json:"payload"` - }{EventID: eventID, EventType: "task.control", DispatcherID: dispatcherID, TenantID: tenantID, Payload: struct { - Status string `json:"status"` - Reason string `json:"reason_message,omitempty"` - }{status, reason}}) + EventID string `json:"event_id"` + EventType string `json:"event_type"` + DispatcherID string `json:"dispatcher_id"` + TenantID int64 `json:"tenant_id"` + Payload controlReceiptPayload `json:"payload"` + }{EventID: eventID, EventType: "task.control", DispatcherID: dispatcherID, TenantID: tenantID, Payload: payload}) if err != nil { return fmt.Errorf("encode task control acknowledgment: %w", err) } if err := contract.ValidateCurrent("mq", body); err != nil { return fmt.Errorf("task control acknowledgment violates MQ contract: %w", err) } - var oldType string - var oldBody []byte - err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&oldType, &oldBody) - if errors.Is(err, sql.ErrNoRows) { + if oldBody == nil { if _, err := tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?)`, dispatcherID, eventID, "task.control", route.BindingKey, body); err != nil { return fmt.Errorf("persist task control outbox: %w", err) } return nil } - if err != nil { - return fmt.Errorf("inspect existing task control outbox: %w", err) - } if oldType != "task.control" || string(oldBody) != string(body) { return fmt.Errorf("control event identity %q conflicts with another outbound result", eventID) } diff --git a/internal/store/control_receipt.go b/internal/store/control_receipt.go new file mode 100644 index 0000000..3d8a49e --- /dev/null +++ b/internal/store/control_receipt.go @@ -0,0 +1,79 @@ +package store + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + + "git.ipao.vip/rogee/go-sip/contracts" +) + +type controlReceiptPayload struct { + TaskID string `json:"task_id,omitempty"` + Action string `json:"action,omitempty"` + Status string `json:"status"` + ReasonCode string `json:"reason_code"` + TaskState string `json:"task_state,omitempty"` + ReasonMessage string `json:"reason_message,omitempty"` +} + +// State is read in the same transaction as the receipt. A missing task is +// distinguishable from a database failure; neither request data nor status +// from another tenant/Dispatcher can supply this fact. +func controlReceiptState(tx *sql.Tx, dispatcherID string, tenantID int64, taskID string) (string, error) { + var state string + err := tx.QueryRow(`SELECT CASE WHEN control_state<>'' THEN control_state ELSE status END FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=? AND task_id=?`, dispatcherID, tenantID, taskID).Scan(&state) + if errors.Is(err, sql.ErrNoRows) { + return "", nil + } + if err != nil { + return "", fmt.Errorf("read task control receipt state: %w", err) + } + return state, nil +} + +// Validate each recoverable field against its own shared-schema definition, +// not against the whole invalid request. An invalid sibling cannot erase a +// valid task ID/action, and supplied task_state is never trusted. +func rejectedControlPayload(tx *sql.Tx, dispatcherID string, tenantID int64, raw json.RawMessage, reasonCode, reason string) (controlReceiptPayload, error) { + result := controlReceiptPayload{Status: "rejected", ReasonCode: reasonCode, ReasonMessage: reason} + schema, err := contracts.CompileCurrent("mq-task-control-receipt.schema.json") + if err != nil { + return result, err + } + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + // A non-object payload has no independently parseable fields. + return result, nil + } + for _, key := range []string{"task_id", "action"} { + rawValue, present := fields[key] + if !present { + continue + } + var value any + if err := json.Unmarshal(rawValue, &value); err != nil { + continue + } + if schema.Properties["payload"].Properties[key].Validate(value) != nil { + continue + } + text, ok := value.(string) + if !ok { + return result, fmt.Errorf("shared control receipt %s schema accepted a non-string", key) + } + if key == "task_id" { + result.TaskID = text + } else { + result.Action = text + } + } + if result.TaskID != "" { + result.TaskState, err = controlReceiptState(tx, dispatcherID, tenantID, result.TaskID) + if err != nil { + return result, err + } + } + return result, nil +} diff --git a/internal/store/control_receipt_test.go b/internal/store/control_receipt_test.go new file mode 100644 index 0000000..ddbcaeb --- /dev/null +++ b/internal/store/control_receipt_test.go @@ -0,0 +1,252 @@ +package store + +import ( + "bytes" + "encoding/json" + "errors" + "strings" + "testing" + + "git.ipao.vip/rogee/go-sip/internal/contract" +) + +func controlPayload(t *testing.T, s *Store, eventID string) map[string]any { + t.Helper() + var body []byte + if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, currentDispatcherID, eventID).Scan(&body); err != nil { + t.Fatal(err) + } + if err := contract.ValidateCurrent("mq", body); err != nil { + t.Fatalf("invalid receipt: %s: %v", body, err) + } + var event struct { + Payload map[string]any `json:"payload"` + } + if err := json.Unmarshal(body, &event); err != nil { + t.Fatal(err) + } + return event.Payload +} + +func TestControlReceiptAllActionsReportDurableState(t *testing.T) { + for _, action := range []string{"start", "pause", "resume", "stop", "edit"} { + t.Run(action, func(t *testing.T) { + s := preparedCurrentCallStore(t) + snapshot := currentStoreSnapshot(t) + snapshot.SIP.Trunks = []byte(strings.Replace(string(snapshot.SIP.Trunks), `"max_concurrent_calls":null`, `"max_concurrent_calls":2`, 1)) + state := "running" + var err error + switch action { + case "start": + err = s.CompleteStart(snapshot, "receipt") + case "edit": + if err = s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil { + t.Fatal(err) + } + if err = s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "prior"); err != nil { + t.Fatal(err) + } + _, err = s.CompleteEdit(snapshot, "receipt") + state = "paused" + default: + if action == "resume" { + if err = s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil { + t.Fatal(err) + } + if err = s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "prior"); err != nil { + t.Fatal(err) + } + } + if err = s.PrepareControl(currentDispatcherID, 1001, "task-asr", action); err != nil { + t.Fatal(err) + } + err = s.CompleteControl(currentDispatcherID, 1001, "task-asr", action, "receipt") + if action == "pause" { + state = "paused" + } + if action == "stop" { + state = "stopped" + } + } + if err != nil { + t.Fatal(err) + } + payload := controlPayload(t, s, "receipt") + want := map[string]any{"task_id": "task-asr", "action": action, "status": "applied", "reason_code": "applied", "task_state": state} + if len(payload) != len(want) { + t.Fatalf("success must contain exactly five fields: %+v", payload) + } + for k, v := range want { + if payload[k] != v { + t.Errorf("%s: got %v, want %v", k, payload[k], v) + } + } + }) + } +} + +func TestRejectedControlExtractsValidFieldsIndependently(t *testing.T) { + for _, tc := range []struct{ name, payload, task, action, state string }{ + {"missing timestamp", `{"task_id":"task-asr","action":"pause","reason":"private"}`, "task-asr", "pause", "running"}, + {"invalid sibling", `{"task_id":"task-asr","action":"pause","reason":99}`, "task-asr", "pause", "running"}, + {"overflowing sibling number", `{"task_id":"task-asr","action":"pause","reason":1e1000}`, "task-asr", "pause", "running"}, + {"non-object payload", `["task-asr","pause"]`, "", "", ""}, + {"null task", `{"task_id":null,"action":"pause"}`, "", "pause", ""}, + {"invalid action type", `{"task_id":"task-asr","action":99}`, "task-asr", "", "running"}, + {"unknown action", `{"task_id":"task-asr","action":"invalid"}`, "task-asr", "", "running"}, + {"invalid task type", `{"task_id":99,"action":"pause"}`, "", "pause", ""}, + {"invalid task syntax", `{"task_id":"bad task","action":"pause"}`, "", "pause", ""}, + {"missing task", `{"action":"pause"}`, "", "pause", ""}, + {"unknown task", `{"task_id":"absent","action":"pause"}`, "absent", "pause", ""}, + {"other tenant task", `{"task_id":"foreign-task","action":"pause"}`, "foreign-task", "pause", ""}, + {"fake requested state", `{"task_id":"task-asr","action":"pause","task_state":"stopped"}`, "task-asr", "pause", "running"}, + } { + t.Run(tc.name, func(t *testing.T) { + s := preparedCurrentCallStore(t) + if _, err := s.db.Exec(`INSERT INTO dispatcher_tasks(dispatcher_id,tenant_id,task_id,task_revision,status,present) VALUES(?,2002,'foreign-task',1,'stopped',1)`, currentDispatcherID); err != nil { + t.Fatal(err) + } + raw := []byte(`{"event_id":"rejected-fields","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":` + tc.payload + `}`) + cause := contract.ValidateCurrent("mq", raw) + if cause == nil { + t.Fatal("fixture must be rejected") + } + if err := s.RejectData(currentDispatcherID, raw, cause); err != nil { + t.Fatal(err) + } + payload := controlPayload(t, s, "rejected-fields") + for k, want := range map[string]string{"task_id": tc.task, "action": tc.action, "task_state": tc.state} { + got, present := payload[k] + if want == "" { + if present { + t.Errorf("unverified %s must be omitted: %v", k, got) + } + } else if got != want { + t.Errorf("valid %s lost: %v, want %s", k, got, want) + } + } + if payload["reason_code"] != "invalid_data" || payload["status"] != "rejected" || payload["reason_message"] == "" { + t.Fatalf("missing rejection cause: %+v", payload) + } + var state string + if err := s.db.QueryRow(`SELECT control_state FROM dispatcher_tasks WHERE dispatcher_id=? AND tenant_id=1001 AND task_id='task-asr'`, currentDispatcherID).Scan(&state); err != nil || state != "" { + t.Fatalf("invalid request executed: %q %v", state, err) + } + }) + } +} + +func TestBusinessRejectedControlReportsTaskIdentityAndActualState(t *testing.T) { + for _, tc := range []struct{ name, task, action, state string }{ + {"stopped task", "task-asr", "resume", "stopped"}, + {"unknown task", "absent", "start", ""}, + {"pending pause", "task-asr", "resume", "pausing"}, + } { + t.Run(tc.name, func(t *testing.T) { + s := preparedCurrentCallStore(t) + if tc.state != "" { + if _, err := s.db.Exec(`UPDATE dispatcher_tasks SET control_state=? WHERE dispatcher_id=? AND tenant_id=1001 AND task_id='task-asr'`, tc.state, currentDispatcherID); err != nil { + t.Fatal(err) + } + } + if err := s.RejectControl(currentDispatcherID, 1001, "business-rejection", tc.task, tc.action, "task cannot accept this control"); err != nil { + t.Fatal(err) + } + payload := controlPayload(t, s, "business-rejection") + for k, want := range map[string]string{"task_id": tc.task, "action": tc.action, "status": "rejected", "reason_code": "control_rejected", "reason_message": "task cannot accept this control"} { + if payload[k] != want { + t.Errorf("%s: got %v, want %v", k, payload[k], want) + } + } + if tc.state == "" { + if _, present := payload["task_state"]; present { + t.Fatal("invented unknown task state") + } + } else if payload["task_state"] != tc.state { + t.Fatalf("actual state missing: %+v", payload) + } + }) + } +} + +func TestRejectedControlReplayPreservesOriginalTaskState(t *testing.T) { + s := preparedCurrentCallStore(t) + raw := []byte(`{"event_id":"rejected-repeat","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"pause"}}`) + cause := contract.ValidateCurrent("mq", raw) + if err := s.RejectData(currentDispatcherID, raw, cause); err != nil { + t.Fatal(err) + } + var original []byte + if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='rejected-repeat'`).Scan(&original); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "rejected-repeat"); err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`UPDATE dispatcher_tasks SET control_state='stopped' WHERE dispatcher_id=? AND tenant_id=1001 AND task_id='task-asr'`, currentDispatcherID); err != nil { + t.Fatal(err) + } + if err := s.RejectData(currentDispatcherID, raw, cause); err != nil { + t.Fatal(err) + } + rows, err := s.ListPendingOutbox(currentDispatcherID) + if err != nil || len(rows) != 1 || !bytes.Equal(original, rows[0].Body) { + t.Fatalf("rejected receipt changed on replay: %+v %v", rows, err) + } +} + +func TestRejectedControlStateReadFailureDoesNotPersistFeedback(t *testing.T) { + s := preparedCurrentCallStore(t) + if _, err := s.db.Exec(`ALTER TABLE dispatcher_tasks RENAME COLUMN control_state TO unavailable_control_state`); err != nil { + t.Fatal(err) + } + raw := []byte(`{"event_id":"state-failure","event_type":"task.control","dispatcher_id":"` + currentDispatcherID + `","tenant_id":1001,"payload":{"task_id":"task-asr","action":"pause"}}`) + if err := s.RejectData(currentDispatcherID, raw, errors.New("invalid structure")); err == nil { + t.Fatal("state lookup failure silently omitted task_state") + } + var count int + if err := s.db.QueryRow(`SELECT count(*) FROM dispatcher_outbox`).Scan(&count); err != nil || count != 0 { + t.Fatalf("feedback persisted on state failure: %d %v", count, err) + } +} + +func TestEditReceiptRetransmitsOriginalStateAfterTaskChanges(t *testing.T) { + s := preparedCurrentCallStore(t) + snapshot := currentStoreSnapshot(t) + snapshot.SIP.Trunks = []byte(strings.Replace(string(snapshot.SIP.Trunks), `"max_concurrent_calls":null`, `"max_concurrent_calls":2`, 1)) + if _, err := s.CompleteEdit(snapshot, "edit-repeat"); err != nil { + t.Fatal(err) + } + var original []byte + if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='edit-repeat'`).Scan(&original); err != nil { + t.Fatal(err) + } + if err := s.MarkOutboxConfirmed(currentDispatcherID, "edit-repeat"); err != nil { + t.Fatal(err) + } + if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil { + t.Fatal(err) + } + if err := s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "pause-later"); err != nil { + t.Fatal(err) + } + if _, err := s.CompleteEdit(snapshot, "edit-repeat"); err != nil { + t.Fatal(err) + } + var repeated []byte + if err := s.db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE event_id='edit-repeat'`).Scan(&repeated); err != nil { + t.Fatal(err) + } + if !bytes.Equal(original, repeated) { + t.Fatalf("original receipt changed: %s -> %s", original, repeated) + } + if payload := controlPayload(t, s, "edit-repeat"); payload["task_state"] != "running" { + t.Fatalf("original state lost: %+v", payload) + } + if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil { + t.Fatal(err) + } + if err := s.CompleteControl(currentDispatcherID, 1001, "task-asr", "pause", "edit-repeat"); err == nil || !strings.Contains(err.Error(), "conflicts") { + t.Fatalf("event identity reused for another action: %v", err) + } +} diff --git a/internal/store/rejected_data.go b/internal/store/rejected_data.go index f78ab41..2399c15 100644 --- a/internal/store/rejected_data.go +++ b/internal/store/rejected_data.go @@ -20,10 +20,11 @@ var ErrUnidentifiableData = errors.New("rejected data identity cannot be verifie // rejection. The caller may acknowledge the input only after this succeeds. func (s *Store) RejectData(dispatcherID string, original []byte, cause error) error { var id struct { - EventID string `json:"event_id"` - EventType string `json:"event_type"` - DispatcherID string `json:"dispatcher_id"` - TenantID int64 `json:"tenant_id"` + EventID string `json:"event_id"` + EventType string `json:"event_type"` + DispatcherID string `json:"dispatcher_id"` + TenantID int64 `json:"tenant_id"` + Payload json.RawMessage `json:"payload"` } if json.Unmarshal(original, &id) != nil || id.DispatcherID != dispatcherID || id.EventID == "" || len(id.EventID) > 255 { return ErrUnidentifiableData @@ -58,9 +59,27 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er if cause == nil { return errors.New("rejection requires an actual failure reason") } - payload := map[string]any{"status": "rejected", "reason_message": cause.Error()} + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + var payload any = map[string]any{"status": "rejected", "reason_message": cause.Error()} + if id.EventType == "task.control" { + replayed, err := replayRejectedEvent(tx, dispatcherID, id.EventID, id.EventType, id.TenantID) + if err != nil { + return err + } + if replayed { + return tx.Commit() + } + payload, err = rejectedControlPayload(tx, dispatcherID, id.TenantID, id.Payload, "invalid_data", cause.Error()) + if err != nil { + return err + } + } if id.EventType == "call.execute" { - payload["reason_code"] = nil + payload = map[string]any{"status": "rejected", "reason_message": cause.Error(), "reason_code": nil} } receipt := map[string]any{"event_id": id.EventID, "event_type": id.EventType, "dispatcher_id": dispatcherID, "payload": payload} if id.EventType != "sip.config" { @@ -95,11 +114,6 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er if err != nil { return err } - tx, err := s.db.Begin() - if err != nil { - return err - } - defer tx.Rollback() if id.EventType == "call.execute" { var admitted int if err := tx.QueryRow(`SELECT count(*) FROM dispatcher_inbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID).Scan(&admitted); err != nil {