226 lines
6.8 KiB
Go
226 lines
6.8 KiB
Go
package dispatcher
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.ipao.vip/rogee/go-sip/contracts"
|
|
"git.ipao.vip/rogee/go-sip/internal/contract"
|
|
"git.ipao.vip/rogee/go-sip/internal/mq"
|
|
"git.ipao.vip/rogee/go-sip/internal/store"
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
"git.ipao.vip/rogee/go-sip/internal/testfixture"
|
|
"github.com/google/uuid"
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
func TestLocalMQActiveControlReplyRecovery(t *testing.T) {
|
|
address := os.Getenv("GO_SIP_LOCAL_QUERY_MQ_URL")
|
|
if address == "" {
|
|
t.Skip("dedicated local RabbitMQ vhost not configured")
|
|
}
|
|
endpoint, err := url.Parse(address)
|
|
if err != nil || (endpoint.Hostname() != "localhost" && endpoint.Hostname() != "127.0.0.1" && endpoint.Hostname() != "::1") {
|
|
t.Fatal("loopback RabbitMQ required")
|
|
}
|
|
id, key := uuid.NewString(), "control."+uuid.NewString()
|
|
broker, err := mq.Open(address, id)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer broker.Close()
|
|
queue, err := broker.DeclareTenantQueue(key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
route, err := tenant.NewDispatcherRoute(id, key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
path := filepath.Join(t.TempDir(), "dispatcher.db")
|
|
s, err := store.Open(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { s.Close() }()
|
|
if err := s.BindDispatcherID(id); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := s.SetQuota("global", 1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
d, err := New(s, broker, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
connection, err := amqp.Dial(address)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer connection.Close()
|
|
channel, err := connection.Channel()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer channel.Close()
|
|
defer channel.QueueDelete(route.InboxQueue, false, false, false)
|
|
defer channel.QueueDelete(route.DeadLetterQueue, false, false, false)
|
|
defer channel.QueueUnbind(mq.SaaSQueue, route.OutboundKey, mq.EventExchange, nil)
|
|
if err := channel.Confirm(false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
encode := func(value any) []byte {
|
|
t.Helper()
|
|
raw, err := json.Marshal(value)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return raw
|
|
}
|
|
raw, err := testfixture.Execute()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var command map[string]any
|
|
if err := json.Unmarshal(raw, &command); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
command["dispatcher_id"], command["tenant_key"] = id, key
|
|
raw = encode(command)
|
|
if _, err := d.AcceptCommand(raw, route.InboundKey); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
task, err := d.ReserveTask(key, "reservation-mq-control", []string{"global"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
mockNow := time.Date(2026, 9, 18, 0, 0, 0, 0, time.UTC)
|
|
coordinator := NewAgentCoordinator(func() time.Time { return mockNow })
|
|
agentStatus := agentStatusForLocalFlow()
|
|
client := startMockAgent(t, &agentStatus)
|
|
if err := coordinator.Register("agent-a", client); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
|
defer cancel()
|
|
if _, err := coordinator.Activate(ctx, "agent-a", "cell-a", "boot-a", "epoch-a", 1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if result, err := d.ExecuteReserved(ctx, coordinator, "agent-a", task, raw, "reservation-mq-control", strings.Repeat("a", 64)); err != nil || result.Unknown {
|
|
t.Fatalf("prepare execution: %v", err)
|
|
}
|
|
controlRaw, err := contracts.Files.ReadFile("upstream/" + contract.MQSourceCommit + "/examples/task-control.json")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var control map[string]any
|
|
if err := json.Unmarshal(controlRaw, &control); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
commandID := uuid.NewString()
|
|
control["dispatcher_id"], control["tenant_key"], control["tenant_id"], control["command_id"] = id, key, task.TenantID, commandID
|
|
now := time.Now().UTC()
|
|
control["issued_at"], control["not_after"] = now.Format(time.RFC3339Nano), now.Add(time.Minute).Format(time.RFC3339Nano)
|
|
payload := control["payload"].(map[string]any)
|
|
payload["task_id"], payload["expected_task_revision"] = task.TaskID, task.TaskRevision
|
|
controlRaw = encode(control)
|
|
send := func() {
|
|
t.Helper()
|
|
confirmation, err := channel.PublishWithDeferredConfirmWithContext(ctx, mq.DefaultExchange, route.InboundKey, true, false, amqp.Publishing{DeliveryMode: amqp.Persistent, ContentType: "application/json", MessageId: commandID, Body: controlRaw})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if ack, err := confirmation.WaitContext(ctx); err != nil || !ack {
|
|
t.Fatalf("control confirm: %v", err)
|
|
}
|
|
delivery, ok, err := channel.Get(queue, false)
|
|
if err != nil || !ok {
|
|
t.Fatalf("control missing: %v", err)
|
|
}
|
|
if err := d.AcceptMQMessage(delivery.Body, delivery.RoutingKey); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := delivery.Ack(false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
send()
|
|
if count, err := d.ProcessTaskControls(ctx, lostControlReply{coordinator}, 10); err == nil || count != 0 {
|
|
t.Fatal("lost reply acknowledged as applied")
|
|
}
|
|
// Dispatcher restart after the Agent applied the control but before its reply
|
|
// was persisted must recover the original target and operation, not execute.
|
|
if err := s.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
s, err = store.Open(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
d, err = New(s, broker, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if count, err := d.ProcessTaskControls(ctx, coordinator, 10); err != nil || count != 1 {
|
|
t.Fatalf("control recovery count=%d error=%v", count, err)
|
|
}
|
|
var finalID string
|
|
for attempt := 0; attempt < 2; attempt++ {
|
|
if attempt > 0 {
|
|
send()
|
|
}
|
|
if _, err := d.FlushOutbox(ctx, 10); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
found := false
|
|
for {
|
|
delivery, ok, err := channel.Get(mq.SaaSQueue, false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !ok {
|
|
break
|
|
}
|
|
var event map[string]any
|
|
if err := json.Unmarshal(delivery.Body, &event); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if event["aggregate_id"] == commandID {
|
|
body := event["payload"].(map[string]any)
|
|
if body["status"] == "applied" {
|
|
if delivery.DeliveryMode != amqp.Persistent || delivery.RoutingKey != route.OutboundKey || delivery.MessageId != event["event_id"] {
|
|
t.Fatal("final control receipt lost identity or persistence")
|
|
}
|
|
if finalID != "" && finalID != delivery.MessageId {
|
|
t.Fatal("duplicate control created a new final receipt")
|
|
}
|
|
finalID = delivery.MessageId
|
|
found = true
|
|
}
|
|
}
|
|
if err := delivery.Ack(false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if !found {
|
|
t.Fatal("no final applied control receipt reached SaaS queue")
|
|
}
|
|
}
|
|
if count, err := d.ProcessTaskControls(ctx, coordinator, 10); err != nil || count != 0 {
|
|
t.Fatal("duplicate control changed Agent again")
|
|
}
|
|
var tasks, revision int
|
|
if err := s.DB().QueryRow(`SELECT COUNT(*),MAX(task_revision) FROM tasks`).Scan(&tasks, &revision); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if tasks != 1 || int64(revision) != task.TaskRevision+1 {
|
|
t.Fatalf("duplicate execution/revision: tasks=%d revision=%d", tasks, revision)
|
|
}
|
|
}
|