orbit/adapters/outbox/mysql/http_integration_test.go

package outbox

import (
	"context"
	"encoding/json"
	"io"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"strconv"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
)

func realClient(t *testing.T) (*orbit.Client, string, string) {
	t.Helper()
	base, path := os.Getenv("ORBIT_TEST_SENDER_URL"), os.Getenv("ORBIT_TEST_SENDER_CREDENTIALS_FILE")
	if base == "" || path == "" {
		t.Skip("requires isolated Orbit process")
	}
	if !strings.HasPrefix(base, "http://127.0.0.1:") {
		t.Fatal("sender HTTP test requires loopback")
	}
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal("cannot read sender credential")
	}
	var cred struct {
		Token string `json:"token"`
	}
	if json.Unmarshal(raw, &cred) != nil || cred.Token == "" {
		t.Fatal("invalid sender credential")
	}
	api, err := orbit.NewClient(base, orbit.Options{Token: cred.Token})
	if err != nil {
		t.Fatal(err)
	}
	return api, base, cred.Token
}
func realStore(t *testing.T) (*Store, *orbit.Client, string, string) {
	t.Helper()
	db, _ := integrationDB(t)
	api, base, token := realClient(t)
	ctx, stop := context.WithTimeout(context.Background(), 10*time.Second)
	defer stop()
	id := testID(t)
	queue, err := api.CreateQueue(ctx, "outbox-test-"+id, orbit.CreateQueueRequest{Name: "outbox-" + id, MaxConcurrency: 1, RatePerSecond: 10})
	if err != nil {
		t.Fatal(err)
	}
	s, err := businessStore(db, queue.Queue.NamespaceID, queue.Queue.ID)
	if err != nil {
		t.Fatal(err)
	}
	return s, api, base, token
}
func cancelAccepted(t *testing.T, api *orbit.Client, id string) {
	t.Helper()
	ctx, stop := context.WithTimeout(context.Background(), 5*time.Second)
	defer stop()
	task, err := api.GetTask(ctx, id)
	if err != nil {
		t.Fatal(err)
	}
	if _, err = api.CancelTask(ctx, id, "outbox-test-cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Revision, Reason: "outbox regression completed"}); err != nil {
		t.Fatal(err)
	}
}

func TestIntegrationSenderHTTPReceiptLoss(t *testing.T) {
	s, api, base, token := realStore(t)
	id := enqueue(t, s)
	target, _ := url.Parse(base)
	var drop atomic.Bool
	drop.Store(true)
	var posts atomic.Int32
	committed := make(chan orbit.SubmitTaskResult, 1)
	proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		up := r.Clone(r.Context())
		up.URL.Scheme = target.Scheme
		up.URL.Host = target.Host
		up.Host = target.Host
		up.RequestURI = ""
		resp, err := http.DefaultTransport.RoundTrip(up)
		if err != nil {
			http.Error(w, "upstream unavailable", 502)
			return
		}
		defer resp.Body.Close()
		raw, err := io.ReadAll(io.LimitReader(resp.Body, (1<<20)+1))
		if err != nil {
			http.Error(w, "upstream read failed", 502)
			return
		}
		if r.Method == "POST" && r.URL.Path == "/v1/tasks" {
			posts.Add(1)
			if r.Header.Get("Idempotency-Key") != "outbox-task-"+id {
				t.Error("original key changed")
			}
			if resp.StatusCode >= 200 && resp.StatusCode < 300 && drop.CompareAndSwap(true, false) {
				var envelope struct {
					Data orbit.SubmitTaskResult `json:"data"`
				}
				if json.Unmarshal(raw, &envelope) != nil || !idPattern.MatchString(envelope.Data.Task.ID) {
					t.Error("invalid real receipt")
					http.Error(w, "bad receipt", 502)
					return
				}
				committed <- envelope.Data
				// 先发送成功响应头与不完整正文，避免Transport把无响应头的断线当成可透明重放。
				w.Header().Set("Content-Type", "application/json")
				w.Header().Set("Content-Length", strconv.Itoa(len(raw)))
				w.WriteHeader(resp.StatusCode)
				w.Write(raw[:1])
				w.(http.Flusher).Flush()
				conn, _, err := w.(http.Hijacker).Hijack()
				if err == nil {
					conn.Close()
				}
				return
			}
		}
		for key, values := range resp.Header {
			w.Header()[key] = values
		}
		w.WriteHeader(resp.StatusCode)
		w.Write(raw)
	}))
	defer proxy.Close()
	through, err := orbit.NewClient(proxy.URL, orbit.Options{Token: token})
	if err != nil {
		t.Fatal(err)
	}
	ctx, stop := context.WithTimeout(context.Background(), 20*time.Second)
	defer stop()
	r, err := s.SendOnce(ctx, through)
	if err != nil || r.State != "PENDING" || r.LastCode != "UNCONFIRMED" {
		t.Fatal("unknown remote commit classified as done", r.State, err)
	}
	var original orbit.SubmitTaskResult
	select {
	case original = <-committed:
	case <-ctx.Done():
		t.Fatal("real accepted receipt not intercepted")
	}
	if _, err = api.GetTask(ctx, original.Task.ID); err != nil {
		t.Fatal("remote was not actually committed", err)
	}
	execSQL(t, s.db, `UPDATE sender_outbox SET wake_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND operation_id=?`, s.namespace, id)
	r, err = s.SendOnce(ctx, through)
	if err != nil || r.State != "ACCEPTED" || r.TaskID == nil || *r.TaskID != original.Task.ID || r.SendCount != 2 {
		t.Fatal("original identity not recovered", r.State, err)
	}
	var receipt orbit.SubmitTaskResult
	if json.Unmarshal(r.ReceiptJSON, &receipt) != nil || !receipt.Deduplicated || !strings.Contains(string(receipt.Task.Payload), "9007199254740993") {
		t.Fatal("replayed receipt or precision lost")
	}
	page, err := api.ListTasks(ctx, orbit.ListTasksRequest{QueueID: s.queue})
	if err != nil || len(page.Items) != 1 || posts.Load() != 2 {
		t.Fatal("remote duplicate task", len(page.Items), err)
	}
	// 相同凭据搭配错误的持久化目标命名空间，必须在提交前停止。
	wrong, _ := businessStore(s.db, testID(t), s.queue)
	wrongID := enqueue(t, wrong)
	r, err = wrong.SendOnce(ctx, through)
	if err != nil || r.OperationID != wrongID || r.State != "REVIEW" || r.LastCode != "TARGET_MISMATCH" || posts.Load() != 2 {
		t.Fatal("wrong namespace submitted", r.State, err)
	}
	cancelAccepted(t, api, original.Task.ID)
	t.Log("real HTTP receipt discarded after Orbit commit; original key replay recovered exactly one task; wrong namespace never posted")
}