rogee/go-sip
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve source event IDs across all feedback

Browse Source
This commit is contained in:
rogee
2026-10-10 15:02:30 +08:00
parent b41dc76ac4
commit c2ee8a5a4b
26 changed files with 553 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/contract/event_identity_test.go
+33
View File
@@ -0,0 +1,33 @@
package contract
import (
"encoding/json"
"os"
"path/filepath"
"testing"
)
func TestCurrentCallExamplesRetainOriginalEventIdentity(t *testing.T) {
var originalID string
for _, name := range []string{"mq-execute", "mq-execute-ack", "mq-result-no-recording", "mq-result-uploaded"} {
raw, err := os.ReadFile(filepath.Join("..", "..", "contracts", "schema", "examples", name+".json"))
if err != nil {
t.Fatal(err)
}
if err := ValidateCurrent("mq", raw); err != nil {
t.Fatalf("%s: %v", name, err)
}
var event struct {
EventID string `json:"event_id"`
}
if err := json.Unmarshal(raw, &event); err != nil {
t.Fatal(err)
}
if originalID == "" {
originalID = event.EventID
}
if event.EventID != originalID {
t.Fatalf("%s changed source identity: want=%q got=%q", name, originalID, event.EventID)
}
}
}
internal/dispatcher/control_test.go
+1 -1
View File
@@ -123,7 +123,7 @@ func TestControlPauseDefaultHangupAckAfterAgentDispatch(t *testing.T) {
if err != nil || len(outbox) != 1 || outbox[0].EventType != "task.control" || !strings.Contains(string(outbox[0].Body), `"status":"applied"`) {
t.Fatalf("no durable applied ack: %+v %v", outbox, err)
}
if err := s.MarkOutboxConfirmed(controller.DispatcherID, "pause-1"); err != nil {
if err := s.MarkOutboxConfirmed(controller.DispatcherID, "task.control", "pause-1"); err != nil {
t.Fatal(err)
}
agent.before = nil
internal/dispatcher/event_identity_test.go
+74
View File
@@ -0,0 +1,74 @@
package dispatcher
import (
"bytes"
"context"
"encoding/json"
"errors"
"testing"
)
func TestOriginalEventIDSurvivesExecuteResultAndLostPublishConfirmation(t *testing.T) {
controller, originator, publisher, s := newExecuteFixture(t)
id := "numberID-0000123/Case"
command := executeBody(t, id, "15003164745")
if err := controller.ProcessExecute(context.Background(), command); err != nil {
t.Fatal(err)
}
if len(originator.calls) != 1 || originator.calls[0].EventID != id {
t.Fatalf("execute changed source identity: %+v", originator.calls)
}
if err := controller.FlushOutbox(context.Background()); err != nil {
t.Fatal(err)
}
if err := s.FinishExecute(controller.DispatcherID, id); err != nil {
t.Fatal(err)
}
var example struct {
Payload json.RawMessage `json:"payload"`
}
if err := json.Unmarshal(configExample(t, "mq-result-no-recording"), &example); err != nil {
t.Fatal(err)
}
result, created, err := s.RecordCallResult(controller.DispatcherID, id, example.Payload)
if err != nil || !created || result.EventID != id {
t.Fatalf("result lost source identity: %+v created=%t err=%v", result, created, err)
}
publisher.err = errors.New("injected confirm loss after publishing original result")
if err := controller.FlushOutbox(context.Background()); err == nil {
t.Fatal("confirmation loss hidden")
}
pending, err := s.ListPendingOutbox(controller.DispatcherID)
if err != nil || len(pending) != 1 || pending[0].EventType != "call.execute.result" || pending[0].EventID != id {
t.Fatalf("result lost or execute confirmation undone: %+v %v", pending, err)
}
if err := controller.ProcessExecute(context.Background(), command); err != nil || len(originator.calls) != 1 {
t.Fatalf("MQ redelivery redialed: %d %v", len(originator.calls), err)
}
publisher.err = nil
if err := controller.FlushOutbox(context.Background()); err != nil {
t.Fatal(err)
}
if len(publisher.bodies) != 3 || !bytes.Equal(publisher.bodies[1], publisher.bodies[2]) {
t.Fatalf("result retry changed message: %d publishes", len(publisher.bodies))
}
for i, body := range publisher.bodies {
var event struct {
EventID string `json:"event_id"`
EventType string `json:"event_type"`
}
if err := json.Unmarshal(body, &event); err != nil || event.EventID != id {
t.Fatalf("publish %d changed source ID: %+v %v", i, event, err)
}
want := "call.execute.result"
if i == 0 {
want = "call.execute"
}
if event.EventType != want {
t.Fatalf("publish %d confused feedback type: %+v", i, event)
}
}
if pending, err := s.ListPendingOutbox(controller.DispatcherID); err != nil || len(pending) != 0 {
t.Fatalf("confirmed result still pending: %+v %v", pending, err)
}
}
internal/dispatcher/execute.go
+3 -3
View File
@@ -314,11 +314,11 @@ func (c *ExecuteController) FlushOutbox(ctx context.Context) error {
var failures []error
for _, event := range pending {
if err := c.Publisher.Publish(ctx, "agent-call.saas.v1", event.RoutingKey, event.Body); err != nil {
failures = append(failures, fmt.Errorf("publish event %q to SaaS queue: %w", event.EventID, err))
failures = append(failures, fmt.Errorf("publish event %q/%q to SaaS queue: %w", event.EventType, event.EventID, err))
continue
}
if err := c.Store.MarkOutboxConfirmed(c.DispatcherID, event.EventID); err != nil {
failures = append(failures, fmt.Errorf("persist queue confirmation for %q: %w", event.EventID, err))
if err := c.Store.MarkOutboxConfirmed(c.DispatcherID, event.EventType, event.EventID); err != nil {
failures = append(failures, fmt.Errorf("persist queue confirmation for %q/%q: %w", event.EventType, event.EventID, err))
}
}
return errors.Join(failures...)
internal/dispatcher/runtime_integration_test.go
+54 -4
View File
@@ -114,7 +114,8 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) {
}
}
publish(controlRoute, controlBody(t, "control-example", "pause", "drain"))
publish(taskRoute, executeBody(t, "call-example", "15803300952"))
sourceID := "numberID-runtime-0000123/Case"
publish(taskRoute, executeBody(t, sourceID, "15803300952"))
snapshot := policySnapshot(t)
sipJSON, err := json.Marshal(snapshot.SIP)
@@ -239,7 +240,7 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) {
}
select {
case spec := <-agent.calls:
if spec.EventID != "call-example" || spec.TrunkID != "trunk-mock" || spec.Callee != "15803300952" {
if spec.EventID != sourceID || spec.TrunkID != "trunk-mock" || spec.Callee != "15803300952" {
t.Fatalf("wrong SaaS event callee or dispatch: %+v", spec)
}
case <-finished:
@@ -278,10 +279,59 @@ func TestRuntimeIsolatedControlBacklogExecuteAndSharedResult(t *testing.T) {
t.Fatal(err)
}
}
if seen["control-example"] != "applied" || seen["resume-integration"] != "applied" || seen["call-example"] != "dispatched" {
if seen["control-example"] != "applied" || seen["resume-integration"] != "applied" || seen[sourceID] != "dispatched" {
t.Fatalf("wrong shared SaaS results: %+v", seen)
}
publish(taskRoute, executeBody(t, "call-example", "15803300952"))
// Complete this isolated Mock call and inspect the actual final result in
// the shared RabbitMQ queue: the already-confirmed dispatch receipt and
// final result must have exactly the same source identity.
if err := db.FinishExecute(id, sourceID); err != nil {
t.Fatal(err)
}
var resultExample struct {
Payload map[string]any `json:"payload"`
}
if err := json.Unmarshal(configExample(t, "mq-result-no-recording"), &resultExample); err != nil {
t.Fatal(err)
}
resultExample.Payload["callee"] = "15803300952"
resultPayload, err := json.Marshal(resultExample.Payload)
if err != nil {
t.Fatal(err)
}
originalResult, created, err := db.RecordCallResult(id, sourceID, resultPayload)
if err != nil || !created || originalResult.EventID != sourceID {
t.Fatalf("final result changed original identity: %+v %t %v", originalResult, created, err)
}
resultDeadline := time.After(5 * time.Second)
for {
msg, ok, err := admin.Get(shared, false)
if err != nil {
t.Fatal(err)
}
if ok {
var event struct {
EventID string `json:"event_id"`
EventType string `json:"event_type"`
}
if err := json.Unmarshal(msg.Body, &event); err != nil || event.EventID != sourceID || event.EventType != "call.execute.result" || msg.MessageId != sourceID || msg.DeliveryMode != amqp.Persistent || string(msg.Body) != string(originalResult.Body) {
t.Fatalf("actual MQ result did not retain original identity/body: %+v %v", event, err)
}
if err := msg.Ack(false); err != nil {
t.Fatal(err)
}
break
}
select {
case <-resultDeadline:
t.Fatal("same-ID final result never reached shared queue")
case <-time.After(20 * time.Millisecond):
}
}
if _, created, err := db.RecordCallResult(id, sourceID, resultPayload); err != nil || created {
t.Fatalf("result duplicate became new output: %t %v", created, err)
}
publish(taskRoute, executeBody(t, sourceID, "15803300952"))
select {
case spec := <-agent.calls:
t.Fatalf("redelivery reoriginated call %s", spec.EventID)
internal/mq/broker_integration_test.go
+3 -3
View File
@@ -243,7 +243,7 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) {
// A dead connection must not claim delivery. An explicit new session can
// publish the unchanged pending identity to the SaaS-owned shared queue.
route, _ := tenant.ResultRoute(ids[0])
pending := []byte(strings.Replace(string(resultBody), `"event_id":"call-result-example"`, `"event_id":"call-result-reopened"`, 1))
pending := []byte(strings.Replace(string(resultBody), `"event_id":"call-example"`, `"event_id":"numberID-reopened-0000123"`, 1))
if err := brokers[0].Close(); err != nil {
t.Fatal(err)
}
@@ -260,7 +260,7 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) {
t.Fatalf("original pending identity was not delivered after reconnect: %v", err)
}
delivery, ok, err := admin.Get(shared, false)
if err != nil || !ok || delivery.MessageId != "call-result-reopened" || string(delivery.Body) != string(pending) || delivery.DeliveryMode != amqp.Persistent {
if err != nil || !ok || delivery.MessageId != "numberID-reopened-0000123" || string(delivery.Body) != string(pending) || delivery.DeliveryMode != amqp.Persistent {
t.Fatalf("reconnected result was not received by the SaaS-owned queue with its original identity: ok=%t err=%v", ok, err)
}
if err := delivery.Ack(false); err != nil {
@@ -365,7 +365,7 @@ func TestBrokerSharedResultQueueAndNoConfigure(t *testing.T) {
if err := applied.Ack(false); err != nil {
t.Fatal(err)
}
if err := s.MarkOutboxConfirmed(ids[0], "control-applied-fields"); err != nil {
if err := s.MarkOutboxConfirmed(ids[0], "task.control", "control-applied-fields"); err != nil {
t.Fatal(err)
}
for _, id := range ids {
internal/store/calls.go
+3 -3
View File
@@ -418,7 +418,7 @@ func insertExecuteAck(tx *sql.Tx, dispatcherID, eventID string, tenantID int64,
func requireExecuteAck(tx *sql.Tx, dispatcherID, eventID string) error {
var eventType string
if err := tx.QueryRow(`SELECT event_type FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&eventType); err != nil {
if err := tx.QueryRow(`SELECT event_type FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute' AND event_id=?`, dispatcherID, eventID).Scan(&eventType); err != nil {
return fmt.Errorf("confirmed call %q has no durable acknowledgment: %w", eventID, err)
}
if eventType != "call.execute" {
@@ -520,8 +520,8 @@ func (s *Store) ListPendingOutbox(dispatcherID string) ([]OutboxEvent, error) {
// Confirm means delivery to a bound queue, not application receipt. The row
// remains durable and is never deleted without separate explicit evidence.
func (s *Store) MarkOutboxConfirmed(dispatcherID, eventID string) error {
result, err := s.db.Exec(`UPDATE dispatcher_outbox SET confirmed=1,confirmed_at=? WHERE dispatcher_id=? AND event_id=? AND confirmed=0`, time.Now().UTC().Format(time.RFC3339Nano), dispatcherID, eventID)
func (s *Store) MarkOutboxConfirmed(dispatcherID, eventType, eventID string) error {
result, err := s.db.Exec(`UPDATE dispatcher_outbox SET confirmed=1,confirmed_at=? WHERE dispatcher_id=? AND event_type=? AND event_id=? AND confirmed=0`, time.Now().UTC().Format(time.RFC3339Nano), dispatcherID, eventType, eventID)
if err != nil {
return fmt.Errorf("persist MQ queue confirm: %w", err)
}
internal/store/calls_test.go
+1 -1
View File
@@ -102,7 +102,7 @@ func TestExecuteRejectsOnlyOnceWithoutFinalResult(t *testing.T) {
if err := s.RejectExecute(cmd.DispatcherID, cmd.EventID, "callee not permitted"); err == nil {
t.Fatal("duplicate rejection created another outbox event")
}
if err := s.MarkOutboxConfirmed(currentDispatcherID, outbox[0].EventID); err != nil {
if err := s.MarkOutboxConfirmed(currentDispatcherID, outbox[0].EventType, outbox[0].EventID); err != nil {
t.Fatal(err)
}
if pending, err := s.ListPendingOutbox(currentDispatcherID); err != nil || len(pending) != 0 {
internal/store/control_ack_test.go
+1 -1
View File
@@ -23,7 +23,7 @@ func TestControlAckOnlyAfterDispatchAndInSameTransaction(t *testing.T) {
if err != nil || len(outbox) != 1 || outbox[0].EventType != "task.control" || !strings.Contains(string(outbox[0].Body), `"status":"applied"`) {
t.Fatalf("missing applied control ack: %+v %v", outbox, err)
}
if err := s.MarkOutboxConfirmed(currentDispatcherID, "pause-event-1"); err != nil {
if err := s.MarkOutboxConfirmed(currentDispatcherID, "task.control", "pause-event-1"); err != nil {
t.Fatal(err)
}
// A repeated control must be dispatched again, not suppressed by event ID.
internal/store/control_flow.go
+2 -2
View File
@@ -231,7 +231,7 @@ func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID,
// 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)
err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='task.control' 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)
}
@@ -266,7 +266,7 @@ func enqueueControlAck(tx *sql.Tx, dispatcherID string, tenantID int64, eventID,
if oldType != "task.control" || string(oldBody) != string(body) {
return fmt.Errorf("control event identity %q conflicts with another outbound result", eventID)
}
if _, err := tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID); err != nil {
if _, err := tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_type='task.control' AND event_id=?`, dispatcherID, eventID); err != nil {
return fmt.Errorf("requeue repeated control acknowledgment: %w", err)
}
return nil
internal/store/control_receipt_test.go
+2 -2
View File
@@ -180,7 +180,7 @@ func TestRejectedControlReplayPreservesOriginalTaskState(t *testing.T) {
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 {
if err := s.MarkOutboxConfirmed(currentDispatcherID, "task.control", "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 {
@@ -221,7 +221,7 @@ func TestEditReceiptRetransmitsOriginalStateAfterTaskChanges(t *testing.T) {
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 {
if err := s.MarkOutboxConfirmed(currentDispatcherID, "task.control", "edit-repeat"); err != nil {
t.Fatal(err)
}
if err := s.PrepareControl(currentDispatcherID, 1001, "task-asr", "pause"); err != nil {
internal/store/event_identity_test.go
+245
View File
@@ -0,0 +1,245 @@
package store
import (
"bytes"
"database/sql"
"encoding/json"
"errors"
"testing"
"git.ipao.vip/rogee/go-sip/internal/contract"
)
func TestSourceEventIdentitySurvivesResultAndRestart(t *testing.T) {
for _, id := range []string{"call-numberID-0000123", "result-numberID-0000123", "NumberID:0000123/Case", " 接收-numberID-0000123 "} {
t.Run(id, func(t *testing.T) {
s := preparedCurrentCallStore(t)
cmd := currentResultCall(t, s, id, true)
result, created, err := s.RecordCallResult(cmd.DispatcherID, id, currentResultPayload(t))
if err != nil || !created || result.EventID != id {
t.Fatalf("source event ID changed: want=%q got=%q created=%t err=%v", id, result.EventID, created, err)
}
pending, err := s.ListPendingOutbox(cmd.DispatcherID)
if err != nil || len(pending) != 2 {
t.Fatalf("ack/result missing: %+v %v", pending, err)
}
for _, event := range pending {
var envelope struct {
EventID string `json:"event_id"`
}
if err := json.Unmarshal(event.Body, &envelope); err != nil || envelope.EventID != id || event.EventID != id {
t.Fatalf("feedback changed source ID: %+v %v", event, err)
}
}
path := s.path
if err := s.Close(); err != nil {
t.Fatal(err)
}
restarted, err := Open(path)
if err != nil {
t.Fatal(err)
}
defer restarted.Close()
again, created, err := restarted.RecordCallResult(cmd.DispatcherID, id, currentResultPayload(t))
if err != nil || created || again.EventID != id || !bytes.Equal(again.Body, result.Body) {
t.Fatalf("restart changed original result: %+v created=%t err=%v", again, created, err)
}
if _, created, err := restarted.RecordExecute(cmd); err != nil || created {
t.Fatalf("duplicate command became a new execution: created=%t err=%v", created, err)
}
})
}
}
func TestSourceEventIdentityAllowsIndependentFeedback(t *testing.T) {
s := preparedCurrentCallStore(t)
cmd := currentResultCall(t, s, "numberID-shared-0000123", true)
// Reproduce the required result identity independently of the generator so
// the old two-column primary key also fails before implementation changes.
if _, err := s.db.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?)`, cmd.DispatcherID, cmd.EventID, "call.execute.result", "mock.route", []byte(`{"source":"original"}`)); err != nil {
t.Fatalf("different feedback types cannot share source ID: %v", err)
}
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "call.execute", cmd.EventID); err != nil {
t.Fatal(err)
}
pending, err := s.ListPendingOutbox(cmd.DispatcherID)
if err != nil || len(pending) != 1 || pending[0].EventType != "call.execute.result" {
t.Fatalf("ack confirmation consumed a different feedback: %+v %v", pending, err)
}
}
func TestSourceEventIdentityConfirmationCannotConsumeAnotherType(t *testing.T) {
for _, first := range []string{"call.execute", "call.execute.result"} {
t.Run(first, func(t *testing.T) {
s := preparedCurrentCallStore(t)
cmd := currentResultCall(t, s, "numberID-confirm-0000123", true)
if _, _, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, currentResultPayload(t)); err != nil {
t.Fatal(err)
}
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "task.control", cmd.EventID); err == nil {
t.Fatal("nonexistent feedback falsely confirmed")
}
if pending, err := s.ListPendingOutbox(cmd.DispatcherID); err != nil || len(pending) != 2 {
t.Fatalf("wrong type changed confirmations: %+v %v", pending, err)
}
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, first, cmd.EventID); err != nil {
t.Fatal(err)
}
pending, err := s.ListPendingOutbox(cmd.DispatcherID)
if err != nil || len(pending) != 1 || pending[0].EventType == first || pending[0].EventID != cmd.EventID {
t.Fatalf("one confirmation consumed both feedback types: %+v %v", pending, err)
}
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, first, cmd.EventID); err == nil {
t.Fatal("already-confirmed feedback falsely reported another transition")
}
if err := s.MarkExecuteDispatched(cmd.DispatcherID, cmd.EventID); err != nil {
t.Fatalf("late ack confused final result with execute receipt: %v", err)
}
if pending, err := s.ListPendingOutbox(cmd.DispatcherID); err != nil || len(pending) != 1 {
t.Fatalf("late ack recreated feedback: %+v %v", pending, err)
}
})
}
}
func TestSourceEventIdentityUploadedResultRetainsOriginal(t *testing.T) {
s := preparedCurrentCallStore(t)
cmd := currentResultCall(t, s, "numberID-uploaded-0000123", true)
grant := currentUploadBinding(cmd)
if _, _, err := s.BindRecordingUpload(grant); err != nil {
t.Fatal(err)
}
proof := UploadProof{UploadID: grant.UploadID, RecordingID: grant.RecordingID, StatusCode: 200, SizeBytes: grant.SizeBytes, SHA256: grant.ChecksumSHA256}
result, created, err := s.RecordUploadedCallResult(cmd.DispatcherID, cmd.EventID, currentUploadedResultPayload(t, grant), proof)
if err != nil || !created || result.EventID != cmd.EventID {
t.Fatalf("uploaded result changed source identity: %+v created=%t err=%v", result, created, err)
}
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "call.execute.result", cmd.EventID); err != nil {
t.Fatal(err)
}
again, created, err := s.RecordUploadedCallResult(cmd.DispatcherID, cmd.EventID, currentUploadedResultPayload(t, grant), proof)
if err != nil || created || again.EventID != cmd.EventID || !bytes.Equal(result.Body, again.Body) {
t.Fatalf("upload retry changed original result: %+v created=%t err=%v", again, created, err)
}
if pending, err := s.ListPendingOutbox(cmd.DispatcherID); err != nil || len(pending) != 1 || pending[0].EventType != "call.execute" {
t.Fatalf("uploaded result confirmation consumed execute receipt: %+v %v", pending, err)
}
}
func TestSourceEventIdentityUploadUsesOnlyFinalResult(t *testing.T) {
s := preparedCurrentCallStore(t)
cmd := currentResultCall(t, s, "numberID-recording-0000123", true)
// A pending or confirmed execute receipt must not deny the first recording.
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, "call.execute", cmd.EventID); err != nil {
t.Fatal(err)
}
grant := currentUploadBinding(cmd)
if _, created, err := s.BindRecordingUpload(grant); err != nil || !created {
t.Fatalf("execute receipt was mistaken for final result: created=%t err=%v", created, err)
}
withoutRecording := currentResultCall(t, s, "numberID-no-recording-0000456", true)
result, _, err := s.RecordCallResult(withoutRecording.DispatcherID, withoutRecording.EventID, currentResultPayload(t))
if err != nil || result.EventID != withoutRecording.EventID {
t.Fatalf("final source ID changed: %+v %v", result, err)
}
if _, _, err := s.BindRecordingUpload(currentUploadBinding(withoutRecording)); !errors.Is(err, ErrResultConflict) {
t.Fatalf("new recording accepted after final result: %v", err)
}
if _, err := s.LoadRecordingUpload(withoutRecording.DispatcherID, withoutRecording.EventID); !errors.Is(err, ErrUploadNotFound) {
t.Fatalf("refused grant left partial recording: %v", err)
}
}
func TestSourceEventIdentityRejectionReplayOnlyRequeuesItsType(t *testing.T) {
for _, kind := range []string{"call.execute", "task.control", "sip.config"} {
t.Run(kind, func(t *testing.T) {
s := preparedCurrentCallStore(t)
id := "numberID-rejected-0000123"
raw, err := json.Marshal(map[string]any{"event_id": id, "event_type": kind, "dispatcher_id": currentDispatcherID, "tenant_id": 1001, "payload": map[string]any{}})
if err != nil {
t.Fatal(err)
}
cause := contract.ValidateCurrent("mq", raw)
if cause == nil {
t.Fatal("fixture unexpectedly valid")
}
if err := s.RejectData(currentDispatcherID, raw, cause); err != nil {
t.Fatal(err)
}
pending, err := s.ListPendingOutbox(currentDispatcherID)
if err != nil || len(pending) != 1 || pending[0].EventID != id || pending[0].EventType != kind {
t.Fatalf("rejection changed source identity/type: %+v %v", pending, err)
}
original := bytes.Clone(pending[0].Body)
if err := s.MarkOutboxConfirmed(currentDispatcherID, kind, id); err != nil {
t.Fatal(err)
}
// A different confirmed feedback sharing the identity must stay
// confirmed while the rejected receipt is requested again.
if _, err := s.db.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body,confirmed) VALUES(?,?,?,?,?,1)`, currentDispatcherID, id, "call.execute.result", "mock.route", []byte(`{}`)); err != nil {
t.Fatal(err)
}
for _, replay := range []bool{false, true} {
if replay {
tenantID := int64(1001)
if kind == "sip.config" {
tenantID = 0
}
if rejected, err := s.ReplayRejectedEvent(currentDispatcherID, id, kind, tenantID); err != nil || !rejected {
t.Fatalf("original rejection not replayed: %t %v", rejected, err)
}
} else if err := s.RejectData(currentDispatcherID, raw, cause); err != nil {
t.Fatal(err)
}
pending, err := s.ListPendingOutbox(currentDispatcherID)
if err != nil || len(pending) != 1 || pending[0].EventID != id || pending[0].EventType != kind || !bytes.Equal(original, pending[0].Body) {
t.Fatalf("rejection requeued another feedback or changed original: %+v %v", pending, err)
}
if err := s.MarkOutboxConfirmed(currentDispatcherID, kind, id); err != nil {
t.Fatal(err)
}
}
})
}
}
func TestSourceEventIdentityRejectsVersionThreeWithoutWriting(t *testing.T) {
s := preparedCurrentCallStore(t)
cmd := currentResultCall(t, s, "numberID-old-layout", true)
if _, err := s.db.Exec(`PRAGMA user_version=3`); err != nil {
t.Fatal(err)
}
var schemaBefore string
if err := s.db.QueryRow(`SELECT sql FROM sqlite_master WHERE name='dispatcher_outbox'`).Scan(&schemaBefore); err != nil {
t.Fatal(err)
}
pendingBefore, err := s.ListPendingOutbox(cmd.DispatcherID)
if err != nil {
t.Fatal(err)
}
path := s.path
if err := s.Close(); err != nil {
t.Fatal(err)
}
if reopened, err := Open(path); err == nil {
reopened.Close()
t.Fatal("old layout admitted or silently migrated")
}
db, err := sql.Open("sqlite", path)
if err != nil {
t.Fatal(err)
}
defer db.Close()
var version int
var schemaAfter string
var body []byte
if err := db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil || version != 3 {
t.Fatalf("old version changed: %d %v", version, err)
}
if err := db.QueryRow(`SELECT sql FROM sqlite_master WHERE name='dispatcher_outbox'`).Scan(&schemaAfter); err != nil || schemaBefore != schemaAfter {
t.Fatalf("old schema changed: %v", err)
}
if err := db.QueryRow(`SELECT body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute' AND event_id=?`, cmd.DispatcherID, cmd.EventID).Scan(&body); err != nil || !bytes.Equal(body, pendingBefore[0].Body) {
t.Fatalf("old feedback changed: %v", err)
}
}
internal/store/rejected_data.go
+2 -2
View File
@@ -125,14 +125,14 @@ func (s *Store) RejectData(dispatcherID string, original []byte, cause error) er
}
var oldType string
var old []byte
err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID).Scan(&oldType, &old)
err = tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, id.EventType, id.EventID).Scan(&oldType, &old)
if errors.Is(err, sql.ErrNoRows) {
_, err = tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?)`, dispatcherID, id.EventID, id.EventType, route.BindingKey, body)
} else if err == nil {
if oldType != id.EventType || !bytes.Equal(old, body) {
return errors.New("rejection conflicts with original event result")
}
_, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, id.EventID)
_, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, id.EventType, id.EventID)
}
if err != nil {
return fmt.Errorf("persist rejected-data feedback: %w", err)
internal/store/rejection_identity.go
+2 -2
View File
@@ -28,7 +28,7 @@ func (s *Store) ReplayRejectedEvent(dispatcherID, eventID, eventType string, ten
func replayRejectedEvent(tx *sql.Tx, dispatcherID, eventID, eventType string, tenantID int64) (bool, error) {
var kind string
var body []byte
err := tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&kind, &body)
err := tx.QueryRow(`SELECT event_type,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, eventType, eventID).Scan(&kind, &body)
if errors.Is(err, sql.ErrNoRows) {
return false, nil
}
@@ -53,6 +53,6 @@ func replayRejectedEvent(tx *sql.Tx, dispatcherID, eventID, eventType string, te
if receipt.Payload.Status != "rejected" {
return false, nil
}
_, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID)
_, err = tx.Exec(`UPDATE dispatcher_outbox SET confirmed=0,confirmed_at=NULL WHERE dispatcher_id=? AND event_type=? AND event_id=?`, dispatcherID, eventType, eventID)
return err == nil, err
}
internal/store/result.go
+3 -9
View File
@@ -2,7 +2,6 @@ package store
import (
"bytes"
"crypto/sha256"
"database/sql"
"encoding/json"
"errors"
@@ -19,11 +18,6 @@ var ErrResultConflict = errors.New("call already has a different final result")
var ErrEndUnconfirmed = errors.New("final result requires confirmed call end and its frozen snapshot")
var ErrResultInvalid = errors.New("final result input violates the approved contract")
func currentResultEventID(dispatcherID, sourceEventID string) string {
identity := sha256.Sum256([]byte("call.execute.result\x00" + dispatcherID + "\x00" + sourceEventID))
return fmt.Sprintf("result-%x", identity[:])
}
// RecordCallResult stores one no-recording final result only after confirmed
// call end. A durable upload grant cannot be bypassed with an empty recording.
func (s *Store) RecordCallResult(dispatcherID, sourceEventID string, payload []byte) (OutboxEvent, bool, error) {
@@ -118,7 +112,7 @@ func (s *Store) recordCallResult(dispatcherID, sourceEventID string, payload []b
if err != nil {
return OutboxEvent{}, false, err
}
eventID := currentResultEventID(dispatcherID, sourceEventID)
eventID := sourceEventID
body, err := json.Marshal(struct {
EventID string `json:"event_id"`
EventType string `json:"event_type"`
@@ -133,7 +127,7 @@ func (s *Store) recordCallResult(dispatcherID, sourceEventID string, payload []b
if err := contract.ValidateCurrent("mq", body); err != nil {
return OutboxEvent{}, false, fmt.Errorf("%w: MQ contract: %v", ErrResultInvalid, err)
}
inserted, err := tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?) ON CONFLICT(dispatcher_id,event_id) DO NOTHING`, dispatcherID, eventID, "call.execute.result", route.BindingKey, body)
inserted, err := tx.Exec(`INSERT INTO dispatcher_outbox(dispatcher_id,event_id,event_type,routing_key,body) VALUES(?,?,?,?,?) ON CONFLICT(dispatcher_id,event_type,event_id) DO NOTHING`, dispatcherID, eventID, "call.execute.result", route.BindingKey, body)
if err != nil {
return OutboxEvent{}, false, fmt.Errorf("persist final result outbox: %w", err)
}
@@ -143,7 +137,7 @@ func (s *Store) recordCallResult(dispatcherID, sourceEventID string, payload []b
}
var stored OutboxEvent
stored.EventID = eventID
if err := tx.QueryRow(`SELECT event_type,routing_key,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, dispatcherID, eventID).Scan(&stored.EventType, &stored.RoutingKey, &stored.Body); err != nil {
if err := tx.QueryRow(`SELECT event_type,routing_key,body FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute.result' AND event_id=?`, dispatcherID, eventID).Scan(&stored.EventType, &stored.RoutingKey, &stored.Body); err != nil {
return OutboxEvent{}, false, fmt.Errorf("inspect existing final result: %w", err)
}
if stored.EventType != "call.execute.result" || stored.RoutingKey != route.BindingKey || !bytes.Equal(stored.Body, body) {
internal/store/result_test.go
+5 -5
View File
@@ -63,7 +63,7 @@ func TestFinalResultNeedsConfirmedEndAndRemainsExactlyOne(t *testing.T) {
t.Fatalf("confirmed end did not release the call before OSS/result delivery: %+v err=%v", occupancy, err)
}
event, created, err := s.RecordCallResult(cmd.DispatcherID, cmd.EventID, payload)
if err != nil || !created || event.EventType != "call.execute.result" || event.EventID == cmd.EventID {
if err != nil || !created || event.EventType != "call.execute.result" || event.EventID != cmd.EventID {
t.Fatalf("exact final result not persisted: event=%+v created=%t err=%v", event, created, err)
}
route, err := tenant.ResultRoute(cmd.DispatcherID)
@@ -78,8 +78,8 @@ func TestFinalResultNeedsConfirmedEndAndRemainsExactlyOne(t *testing.T) {
if err != nil || created || !bytes.Equal(repeated.Body, event.Body) {
t.Fatalf("redelivered identical result inserted twice: created=%t err=%v", created, err)
}
for _, id := range []string{cmd.EventID, event.EventID} {
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, id); err != nil {
for _, feedback := range outbox {
if err := s.MarkOutboxConfirmed(cmd.DispatcherID, feedback.EventType, feedback.EventID); err != nil {
t.Fatal(err)
}
}
@@ -124,7 +124,7 @@ func TestFinalResultAfterEndRacedOriginateAck(t *testing.T) {
t.Fatalf("late originate response conflicted with the confirmed result: %v", err)
}
outbox, err := s.ListPendingOutbox(cmd.DispatcherID)
if err != nil || len(outbox) != 2 || outbox[0].EventID != cmd.EventID || outbox[1].EventID != result.EventID {
if err != nil || len(outbox) != 2 || outbox[0].EventID != cmd.EventID || outbox[0].EventType != "call.execute" || outbox[1].EventID != cmd.EventID || outbox[1].EventType != "call.execute.result" || result.EventID != cmd.EventID {
t.Fatalf("early end lost its acknowledgment or produced duplicate results: outbox=%+v err=%v", outbox, err)
}
}
@@ -220,7 +220,7 @@ func TestFinalResultRestartResumesSameOutboxBody(t *testing.T) {
if err != nil || created || repeated.EventID != original.EventID || !bytes.Equal(repeated.Body, original.Body) {
t.Fatalf("retry created a second result after restart: created=%t err=%v", created, err)
}
if err := restarted.MarkOutboxConfirmed(cmd.DispatcherID, original.EventID); err != nil {
if err := restarted.MarkOutboxConfirmed(cmd.DispatcherID, original.EventType, original.EventID); err != nil {
t.Fatal(err)
}
pending, err = restarted.ListPendingOutbox(cmd.DispatcherID)
internal/store/store.go
+10 -3
View File
@@ -83,7 +83,7 @@ CREATE TABLE IF NOT EXISTS dispatcher_outbox (
body BLOB NOT NULL,
confirmed INTEGER NOT NULL DEFAULT 0 CHECK(confirmed IN (0,1)),
confirmed_at TEXT,
PRIMARY KEY(dispatcher_id,event_id)
PRIMARY KEY(dispatcher_id,event_type,event_id)
);
CREATE TABLE IF NOT EXISTS dispatcher_recordings (
dispatcher_id TEXT NOT NULL,
@@ -155,9 +155,16 @@ func Open(path string) (_ *Store, err error) {
return nil, fmt.Errorf("unknown SQLite layout version %d without tables: preserve database before admission", version)
}
} else {
if len(seen) != len(allowed) || version != 3 {
if len(seen) != len(allowed) || version != 4 {
return nil, fmt.Errorf("existing SQLite layout is incomplete or obsolete (tables=%d, version=%d): preserve data before admission", len(seen), version)
}
var outboxIdentityColumns int
if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('dispatcher_outbox') WHERE (name='dispatcher_id' AND pk=1) OR (name='event_type' AND pk=2) OR (name='event_id' AND pk=3)`).Scan(&outboxIdentityColumns); err != nil {
return nil, fmt.Errorf("inspect typed feedback identity layout: %w", err)
}
if outboxIdentityColumns != 3 {
return nil, errors.New("existing SQLite lacks typed feedback identities: preserve data before admission")
}
var requiredColumns int
if err := db.QueryRow(`SELECT COUNT(*) FROM pragma_table_info('dispatcher_state') WHERE name IN ('discovery_ready','applied_sip_revision','pending_sip_revision')`).Scan(&requiredColumns); err != nil {
return nil, fmt.Errorf("inspect durable SIP state layout: %w", err)
@@ -194,7 +201,7 @@ func Open(path string) (_ *Store, err error) {
if _, err := tx.Exec(schema); err != nil {
return nil, fmt.Errorf("create current SQLite schema: %w", err)
}
if _, err := tx.Exec(`PRAGMA user_version=3`); err != nil {
if _, err := tx.Exec(`PRAGMA user_version=4`); err != nil {
return nil, fmt.Errorf("set current SQLite layout version: %w", err)
}
if err := tx.Commit(); err != nil {
internal/store/upload.go
+1 -1
View File
@@ -107,7 +107,7 @@ func (s *Store) BindRecordingUpload(proposed RecordingGrant) (RecordingGrant, bo
return RecordingGrant{}, false, ErrUploadConflict
}
var finalized int
err = tx.QueryRow(`SELECT 1 FROM dispatcher_outbox WHERE dispatcher_id=? AND event_id=?`, proposed.DispatcherID, currentResultEventID(proposed.DispatcherID, proposed.SourceEventID)).Scan(&finalized)
err = tx.QueryRow(`SELECT 1 FROM dispatcher_outbox WHERE dispatcher_id=? AND event_type='call.execute.result' AND event_id=?`, proposed.DispatcherID, proposed.SourceEventID).Scan(&finalized)
if err == nil {
return RecordingGrant{}, false, ErrResultConflict
}
internal/store/upload_test.go
+1 -1
View File
@@ -55,7 +55,7 @@ func TestUploadBindingPersistsOnlyOneOriginalAssetAcrossRestart(t *testing.T) {
}
var version, seq int
var databaseName, databasePath string
if err := s.db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil || version != 3 {
if err := s.db.QueryRow(`PRAGMA user_version`).Scan(&version); err != nil || version != 4 {
t.Fatalf("current schema is not the new, reviewed recording layout: version=%d err=%v", version, err)
}
if err := s.db.QueryRow(`PRAGMA database_list`).Scan(&seq, &databaseName, &databasePath); err != nil || databasePath == "" {