183 lines
6.3 KiB
Go
183 lines
6.3 KiB
Go
package rpc_test
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"net/url"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
agentv1 "git.ipao.vip/rogee/go-sip/gen/agent/v1"
|
|
"git.ipao.vip/rogee/go-sip/internal/dispatcher"
|
|
"git.ipao.vip/rogee/go-sip/internal/mq"
|
|
ossclient "git.ipao.vip/rogee/go-sip/internal/oss"
|
|
"git.ipao.vip/rogee/go-sip/internal/rpc"
|
|
"git.ipao.vip/rogee/go-sip/internal/store"
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
"github.com/google/uuid"
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
)
|
|
|
|
type lostPublicationResult struct{ broker *mq.Broker }
|
|
|
|
func (p lostPublicationResult) Publish(ctx context.Context, exchange, key string, body []byte) error {
|
|
if err := p.broker.Publish(ctx, exchange, key, body); err != nil {
|
|
return err
|
|
}
|
|
return errors.New("injected loss of publication result")
|
|
}
|
|
|
|
func TestLocalUploadNoticeSurvivesUnroutableAndRestart(t *testing.T) {
|
|
address := os.Getenv("GO_SIP_LOCAL_UPLOAD_MQ_URL")
|
|
if address == "" {
|
|
t.Skip("dedicated local upload RabbitMQ vhost not configured")
|
|
}
|
|
u, err := url.Parse(address)
|
|
if err != nil || (u.Hostname() != "127.0.0.1" && u.Hostname() != "::1" && u.Hostname() != "localhost") {
|
|
t.Fatal("loopback RabbitMQ required")
|
|
}
|
|
id, key := uuid.NewString(), "upload."+uuid.NewString()
|
|
broker, err := mq.Open(address, id)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer broker.Close()
|
|
if _, err := broker.DeclareTenantQueue(key); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
route, err := tenant.NewDispatcherRoute(id, key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
conn, err := amqp.Dial(address)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
channel, err := conn.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.QueueUnbind(mq.SaaSQueue, route.OutboundKey, mq.EventExchange, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
path := filepath.Join(t.TempDir(), "dispatcher.db")
|
|
st, err := store.Open(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { st.Close() }()
|
|
if err := st.BindDispatcherID(id); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
signer, err := ossclient.NewClient(ossclient.Config{Endpoint: "https://oss.invalid", Region: "cn-beijing", Bucket: "local-test", AccessKeyID: "local-test", AccessKeySecret: "local-test", GrantTTL: 15 * time.Minute, MaxAssetBytes: 1024})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
server, err := rpc.NewDispatcherUploadServer(st, signer, time.Now, false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
meta := &agentv1.RequestMeta{AgentId: "agent-a", CellId: "cell-a", OperationId: "op-a", IdempotencyKey: "key-a", TraceId: "trace-a"}
|
|
binding := &agentv1.ExecutionBinding{TenantId: "tenant-a", TenantKey: key, TaskId: "task-a", ExecutionId: "execution-a", CallId: "call-a"}
|
|
asset := &agentv1.AssetDescriptor{Kind: agentv1.AssetKind_ASSET_KIND_RECORDING, AssetId: "recording-a", CallId: "call-a", Format: "wav", Channels: 1, SampleRateHz: 8000, DurationMs: 10, SizeBytes: 4, ChecksumSha256: strings.Repeat("a", 64)}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
grant, err := server.RequestUpload(ctx, &agentv1.RequestUploadRequest{Meta: meta, Binding: binding, Asset: asset, UploadId: "upload-a"})
|
|
if err != nil || grant.GetGrant() == nil {
|
|
t.Fatalf("grant: %v", err)
|
|
}
|
|
request := &agentv1.CompleteUploadRequest{Meta: meta, Binding: binding, Asset: asset, UploadId: "upload-a", UploadedSizeBytes: 4, UploadedChecksumSha256: asset.ChecksumSha256}
|
|
if _, err := server.CompleteUpload(ctx, request); status.Code(err) != codes.Unavailable {
|
|
t.Fatalf("premature completion: %v", err)
|
|
}
|
|
var original []byte
|
|
var eventID string
|
|
if err := st.DB().QueryRow(`SELECT event_id,body FROM outbox`).Scan(&eventID, &original); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
d, err := dispatcher.New(st, broker, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := d.FlushOutbox(ctx, 10); err == nil {
|
|
t.Fatal("unroutable notice counted as delivery")
|
|
}
|
|
if record, err := st.LoadUpload("upload-a"); err != nil || record.State != "uploaded" {
|
|
t.Fatalf("upload fact lost after return: %v", err)
|
|
}
|
|
if err := st.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
st, err = store.Open(path)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.BindDispatcherID(id); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := st.RecoverOutbox(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := channel.QueueBind(mq.SaaSQueue, route.OutboundKey, mq.EventExchange, false, nil); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// Model the application losing the publication result after RabbitMQ has
|
|
// accepted the persistent message. This is fault injection, not a simulated
|
|
// broker protocol failure: the underlying publication is real.
|
|
d, err = dispatcher.New(st, lostPublicationResult{broker}, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := d.FlushOutbox(ctx, 10); err == nil {
|
|
t.Fatal("lost confirmation incorrectly completed delivery")
|
|
}
|
|
if record, err := st.LoadUpload("upload-a"); err != nil || record.State != "uploaded" {
|
|
t.Fatal("unknown publication result was treated as complete")
|
|
}
|
|
d, err = dispatcher.New(st, broker, nil)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if n, err := d.FlushOutbox(ctx, 10); err != nil || n != 1 {
|
|
t.Fatalf("recovery: count=%d err=%v", n, err)
|
|
}
|
|
// No SaaS consumer exists. Delivery completes on queue publication alone.
|
|
server, err = rpc.NewDispatcherUploadServer(st, signer, time.Now, false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
response, err := server.CompleteUpload(ctx, request)
|
|
if err != nil || response.GetState() != agentv1.UploadState_UPLOAD_STATE_COMPLETED {
|
|
t.Fatalf("completion: %v", err)
|
|
}
|
|
for delivery := 0; delivery < 2; delivery++ {
|
|
message, ok, err := channel.Get(mq.SaaSQueue, false)
|
|
if err != nil || !ok {
|
|
t.Fatalf("queue reception: %v", err)
|
|
}
|
|
if message.DeliveryMode != amqp.Persistent || message.MessageId != eventID || !bytes.Equal(message.Body, original) {
|
|
t.Fatal("original persistent notification identity/body was not retained")
|
|
}
|
|
if err := message.Ack(false); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if _, err := server.CompleteUpload(ctx, request); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if n, err := d.FlushOutbox(ctx, 10); err != nil || n != 0 {
|
|
t.Fatalf("duplicate completion republished: %d %v", n, err)
|
|
}
|
|
}
|