240 lines
6.6 KiB
Go
240 lines
6.6 KiB
Go
package outbox
|
||
|
||
import (
|
||
"bufio"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"net/http"
|
||
"os"
|
||
"os/exec"
|
||
"strings"
|
||
"syscall"
|
||
"testing"
|
||
"time"
|
||
|
||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||
)
|
||
|
||
type roundTripper func(*http.Request) (*http.Response, error)
|
||
|
||
func (f roundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
|
||
|
||
type senderMarker struct {
|
||
Phase string `json:"phase"`
|
||
TaskID string `json:"task_id"`
|
||
Sends uint32 `json:"sends"`
|
||
Duplicate bool `json:"duplicate"`
|
||
}
|
||
|
||
func emitMarker(m senderMarker) {
|
||
raw, _ := json.Marshal(m)
|
||
fmt.Println("ORBIT_SENDER_TEST " + string(raw))
|
||
}
|
||
|
||
// 只存在于测试二进制，使用正式Store和真实HTTP；屏障位于平台提交后、本地确认前。
|
||
func TestSenderProcessHelper(t *testing.T) {
|
||
mode := os.Getenv("ORBIT_TEST_SENDER_CHILD")
|
||
if mode == "" {
|
||
t.Skip("subprocess helper")
|
||
}
|
||
if mode != "hold_after_remote_commit" && mode != "recover" {
|
||
t.Fatal("invalid child mode")
|
||
}
|
||
db, _ := integrationDB(t)
|
||
_, base, token := realClient(t)
|
||
s, err := businessStore(db, os.Getenv("ORBIT_TEST_SENDER_NAMESPACE"), os.Getenv("ORBIT_TEST_SENDER_QUEUE"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
options := orbit.Options{Token: token}
|
||
if mode == "hold_after_remote_commit" {
|
||
options.HTTPClient = &http.Client{Timeout: 10 * time.Second, Transport: roundTripper(func(r *http.Request) (*http.Response, error) {
|
||
resp, err := http.DefaultTransport.RoundTrip(r)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if r.Method != "POST" || r.URL.Path != "/v1/tasks" {
|
||
return resp, nil
|
||
}
|
||
defer resp.Body.Close()
|
||
raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
var envelope struct {
|
||
Data orbit.SubmitTaskResult `json:"data"`
|
||
}
|
||
if resp.StatusCode != 201 || json.Unmarshal(raw, &envelope) != nil || !idPattern.MatchString(envelope.Data.Task.ID) {
|
||
return nil, errors.New("child did not receive actual first acceptance")
|
||
}
|
||
emitMarker(senderMarker{Phase: "remote_committed", TaskID: envelope.Data.Task.ID})
|
||
<-r.Context().Done()
|
||
return nil, r.Context().Err()
|
||
})}
|
||
}
|
||
api, err := orbit.NewClient(base, options)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
||
defer stop()
|
||
for ctx.Err() == nil {
|
||
r, err := s.SendOnce(ctx, api)
|
||
if err == nil {
|
||
if r.State != "ACCEPTED" {
|
||
t.Fatal("child did not accept", r.State)
|
||
}
|
||
var receipt orbit.SubmitTaskResult
|
||
if json.Unmarshal(r.ReceiptJSON, &receipt) != nil || !receipt.Deduplicated {
|
||
t.Fatal("child did not recover original receipt")
|
||
}
|
||
emitMarker(senderMarker{Phase: "accepted", TaskID: *r.TaskID, Sends: r.SendCount, Duplicate: receipt.Deduplicated})
|
||
return
|
||
}
|
||
if !errors.Is(err, ErrIdle) {
|
||
t.Fatal(err)
|
||
}
|
||
timer := time.NewTimer(100 * time.Millisecond)
|
||
select {
|
||
case <-ctx.Done():
|
||
timer.Stop()
|
||
case <-timer.C:
|
||
}
|
||
}
|
||
t.Fatal("child sender timed out")
|
||
}
|
||
|
||
type senderChild struct {
|
||
cmd *exec.Cmd
|
||
done chan struct{}
|
||
err error
|
||
markers chan senderMarker
|
||
}
|
||
|
||
func startSender(t *testing.T, s *Store, mode string) *senderChild {
|
||
t.Helper()
|
||
binary, err := os.Executable()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cmd := exec.Command(binary, "-test.run=^TestSenderProcessHelper$")
|
||
values := map[string]string{"ORBIT_TEST_SENDER_CHILD": mode, "ORBIT_TEST_SENDER_NAMESPACE": s.namespace, "ORBIT_TEST_SENDER_QUEUE": s.queue}
|
||
for _, entry := range os.Environ() {
|
||
key, _, _ := strings.Cut(entry, "=")
|
||
if _, ok := values[key]; !ok {
|
||
cmd.Env = append(cmd.Env, entry)
|
||
}
|
||
}
|
||
for key, value := range values {
|
||
cmd.Env = append(cmd.Env, key+"="+value)
|
||
}
|
||
stdout, err := cmd.StdoutPipe()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cmd.Stderr = io.Discard
|
||
p := &senderChild{cmd: cmd, done: make(chan struct{}), markers: make(chan senderMarker, 4)}
|
||
if err = cmd.Start(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
go func() {
|
||
scanner := bufio.NewScanner(stdout)
|
||
for scanner.Scan() {
|
||
if raw, ok := strings.CutPrefix(scanner.Text(), "ORBIT_SENDER_TEST "); ok {
|
||
var m senderMarker
|
||
if json.Unmarshal([]byte(raw), &m) == nil {
|
||
p.markers <- m
|
||
}
|
||
}
|
||
}
|
||
p.err = cmd.Wait()
|
||
close(p.done)
|
||
}()
|
||
t.Cleanup(func() {
|
||
select {
|
||
case <-p.done:
|
||
return
|
||
default:
|
||
}
|
||
_ = cmd.Process.Kill()
|
||
select {
|
||
case <-p.done:
|
||
case <-time.After(5 * time.Second):
|
||
t.Error("owned sender child did not exit")
|
||
}
|
||
})
|
||
return p
|
||
}
|
||
func (p *senderChild) marker(t *testing.T, ctx context.Context, phase string) senderMarker {
|
||
t.Helper()
|
||
for {
|
||
select {
|
||
case m := <-p.markers:
|
||
if m.Phase == phase {
|
||
return m
|
||
}
|
||
case <-p.done:
|
||
select {
|
||
case m := <-p.markers:
|
||
if m.Phase == phase {
|
||
return m
|
||
}
|
||
default:
|
||
}
|
||
t.Fatal("sender exited without expected marker", p.err)
|
||
case <-ctx.Done():
|
||
t.Fatal("sender marker timeout")
|
||
}
|
||
}
|
||
}
|
||
func TestIntegrationSenderProcessCommitRecovery(t *testing.T) {
|
||
s, api, _, _ := realStore(t)
|
||
id := enqueue(t, s)
|
||
ctx, stop := context.WithTimeout(context.Background(), 50*time.Second)
|
||
defer stop()
|
||
first := startSender(t, s, "hold_after_remote_commit")
|
||
committed := first.marker(t, ctx, "remote_committed")
|
||
if _, err := api.GetTask(ctx, committed.TaskID); err != nil {
|
||
t.Fatal("remote commit not confirmed", err)
|
||
}
|
||
r := status(t, s, id)
|
||
if r.State != "SENDING" || r.TaskID != nil || r.SendCount != 1 {
|
||
t.Fatal("local receipt already confirmed")
|
||
}
|
||
if err := first.cmd.Process.Kill(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
select {
|
||
case <-first.done:
|
||
case <-ctx.Done():
|
||
t.Fatal("sender kill not observed")
|
||
}
|
||
state, ok := first.cmd.ProcessState.Sys().(syscall.WaitStatus)
|
||
if !ok || !state.Signaled() || state.Signal() != syscall.SIGKILL {
|
||
t.Fatal("sender was not killed", first.err)
|
||
}
|
||
restarted := startSender(t, s, "recover")
|
||
result := restarted.marker(t, ctx, "accepted")
|
||
select {
|
||
case <-restarted.done:
|
||
case <-ctx.Done():
|
||
t.Fatal("replacement sender not exited")
|
||
}
|
||
if restarted.err != nil || result.TaskID != committed.TaskID || result.Sends != 2 || !result.Duplicate {
|
||
t.Fatal("replacement did not recover original receipt", restarted.err)
|
||
}
|
||
r = status(t, s, id)
|
||
if r.State != "ACCEPTED" || r.TaskID == nil || *r.TaskID != committed.TaskID {
|
||
t.Fatal("durable local acceptance missing")
|
||
}
|
||
page, err := api.ListTasks(ctx, orbit.ListTasksRequest{QueueID: s.queue})
|
||
if err != nil || len(page.Items) != 1 || countRows(t, s, "sender_jobs") != 1 {
|
||
t.Fatal("restart duplicated remote task or business record", err)
|
||
}
|
||
cancelAccepted(t, api, committed.TaskID)
|
||
t.Log("sender SIGKILL observed after actual Orbit commit, before local confirmation; new process waited for natural lease expiry and recovered one task using original key")
|
||
}
|