299 lines
8.9 KiB
Go
299 lines
8.9 KiB
Go
package inbox
|
||
|
||
import (
|
||
"bufio"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"os"
|
||
"os/exec"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
|
||
orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
|
||
)
|
||
|
||
type receiverMarker struct {
|
||
Phase string `json:"phase"`
|
||
EventID string `json:"event_id"`
|
||
Duplicate bool `json:"duplicate"`
|
||
}
|
||
|
||
func receiverClient(t *testing.T) *orbit.Client {
|
||
t.Helper()
|
||
base, path := os.Getenv("ORBIT_TEST_RECEIVER_URL"), os.Getenv("ORBIT_TEST_RECEIVER_CREDENTIALS_FILE")
|
||
if base == "" || path == "" {
|
||
t.Skip("requires isolated real Orbit process and receiver credential")
|
||
}
|
||
if !strings.HasPrefix(base, "http://127.0.0.1:") {
|
||
t.Fatal("receiver process integration requires loopback Orbit")
|
||
}
|
||
raw, err := os.ReadFile(path)
|
||
if err != nil {
|
||
t.Fatal("cannot read test credential")
|
||
}
|
||
var credentials struct {
|
||
Token string `json:"token"`
|
||
}
|
||
if json.Unmarshal(raw, &credentials) != nil || credentials.Token == "" {
|
||
t.Fatal("invalid test credential")
|
||
}
|
||
c, err := orbit.NewClient(base, orbit.Options{Token: credentials.Token})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return c
|
||
}
|
||
|
||
// 此入口只存在于测试二进制。处理器使用正式Store和Consumer，故障屏障位于Apply提交后、返回nil前。
|
||
func TestReceiverProcessHelper(t *testing.T) {
|
||
mode := os.Getenv("ORBIT_TEST_RECEIVER_CHILD")
|
||
if mode == "" {
|
||
t.Skip("subprocess helper")
|
||
}
|
||
if mode != "hold_after_commit" && mode != "confirm_duplicate" {
|
||
t.Fatal("invalid receiver child mode")
|
||
}
|
||
db, _ := integrationDB(t)
|
||
api := receiverClient(t)
|
||
s, err := New(db, os.Getenv("ORBIT_TEST_RECEIVER_NAME"), os.Getenv("ORBIT_TEST_RECEIVER_QUEUE"))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
||
defer stop()
|
||
consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: os.Getenv("ORBIT_TEST_RECEIVER_SUBSCRIPTION"), Capacity: 1, ShutdownTimeout: time.Second, Handle: func(work context.Context, event orbit.ResultEvent) error {
|
||
result, err := s.Apply(work, event)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
marker, _ := json.Marshal(receiverMarker{Phase: "committed", EventID: event.ID, Duplicate: result.Duplicate})
|
||
fmt.Println("ORBIT_RECEIVER_TEST " + string(marker))
|
||
if mode == "hold_after_commit" {
|
||
<-work.Done()
|
||
return work.Err()
|
||
}
|
||
if !result.Duplicate {
|
||
return errors.New("restarted receiver did not find committed inbox")
|
||
}
|
||
return nil
|
||
}})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
run, cancel := context.WithCancel(ctx)
|
||
defer cancel()
|
||
done := make(chan error, 1)
|
||
go func() { done <- consumer.Run(run) }()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
cancel()
|
||
<-done
|
||
t.Fatal("receiver helper timed out")
|
||
case err := <-done:
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return
|
||
case event, ok := <-consumer.Events():
|
||
if !ok {
|
||
if err := <-done; err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return
|
||
}
|
||
if event.Type == "acknowledged" {
|
||
marker, _ := json.Marshal(receiverMarker{Phase: "acknowledged", EventID: event.EventID})
|
||
fmt.Println("ORBIT_RECEIVER_TEST " + string(marker))
|
||
cancel()
|
||
if err := <-done; err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
type receiverChild struct {
|
||
cmd *exec.Cmd
|
||
done chan struct{}
|
||
waitErr error
|
||
markers chan receiverMarker
|
||
}
|
||
|
||
func receiverEnv(values map[string]string) []string {
|
||
out := []string{}
|
||
for _, entry := range os.Environ() {
|
||
key, _, _ := strings.Cut(entry, "=")
|
||
if _, found := values[key]; !found {
|
||
out = append(out, entry)
|
||
}
|
||
}
|
||
for key, value := range values {
|
||
out = append(out, key+"="+value)
|
||
}
|
||
return out
|
||
}
|
||
func startReceiverChild(t *testing.T, s *Store, sub, mode string) *receiverChild {
|
||
t.Helper()
|
||
binary, err := os.Executable()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cmd := exec.Command(binary, "-test.run=^TestReceiverProcessHelper$")
|
||
cmd.Env = receiverEnv(map[string]string{"ORBIT_TEST_RECEIVER_CHILD": mode, "ORBIT_TEST_RECEIVER_NAME": s.consumer, "ORBIT_TEST_RECEIVER_QUEUE": s.queue, "ORBIT_TEST_RECEIVER_SUBSCRIPTION": sub})
|
||
stdout, err := cmd.StdoutPipe()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cmd.Stderr = io.Discard
|
||
p := &receiverChild{cmd: cmd, done: make(chan struct{}), markers: make(chan receiverMarker, 8)}
|
||
if err = cmd.Start(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
readDone := make(chan struct{})
|
||
go func() {
|
||
defer close(readDone)
|
||
scanner := bufio.NewScanner(stdout)
|
||
for scanner.Scan() {
|
||
line := scanner.Text()
|
||
if !strings.HasPrefix(line, "ORBIT_RECEIVER_TEST ") {
|
||
continue
|
||
}
|
||
var marker receiverMarker
|
||
if json.Unmarshal([]byte(strings.TrimPrefix(line, "ORBIT_RECEIVER_TEST ")), &marker) == nil {
|
||
select {
|
||
case p.markers <- marker:
|
||
default:
|
||
}
|
||
}
|
||
}
|
||
}()
|
||
go func() { <-readDone; p.waitErr = 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 receiver child failed to exit")
|
||
}
|
||
})
|
||
return p
|
||
}
|
||
func (p *receiverChild) marker(t *testing.T, ctx context.Context, phase string) receiverMarker {
|
||
t.Helper()
|
||
for {
|
||
select {
|
||
case marker := <-p.markers:
|
||
if marker.Phase == phase {
|
||
return marker
|
||
}
|
||
case <-p.done:
|
||
select {
|
||
case marker := <-p.markers:
|
||
if marker.Phase == phase {
|
||
return marker
|
||
}
|
||
default:
|
||
}
|
||
t.Fatal("receiver child exited before expected boundary", p.waitErr)
|
||
case <-ctx.Done():
|
||
t.Fatal("receiver child did not reach " + phase)
|
||
}
|
||
}
|
||
}
|
||
func (p *receiverChild) kill(t *testing.T) {
|
||
t.Helper()
|
||
select {
|
||
case <-p.done:
|
||
t.Fatal("receiver exited before fault injection")
|
||
default:
|
||
}
|
||
if err := p.cmd.Process.Kill(); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
select {
|
||
case <-p.done:
|
||
var exit *exec.ExitError
|
||
if !errors.As(p.waitErr, &exit) || exit.ExitCode() != -1 {
|
||
t.Fatal("receiver did not terminate by signal")
|
||
}
|
||
case <-time.After(5 * time.Second):
|
||
t.Fatal("receiver SIGKILL was not confirmed")
|
||
}
|
||
}
|
||
|
||
func TestIntegrationReceiverProcessCommitBeforeAck(t *testing.T) {
|
||
db, _ := integrationDB(t)
|
||
api := receiverClient(t)
|
||
ctx, stop := context.WithTimeout(context.Background(), 45*time.Second)
|
||
defer stop()
|
||
id := newID(t)
|
||
queue, err := api.CreateQueue(ctx, "queue-"+id, orbit.CreateQueueRequest{Name: "inbox-process-" + id, MaxConcurrency: 1, RatePerSecond: 10})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
s, err := New(db, "restart-"+id, queue.Queue.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
created, err := api.CreateSubscription(ctx, "sub-"+id, orbit.CreateSubscriptionRequest{Name: "inbox-process-" + id, QueueID: queue.Queue.ID, Types: []string{"task.cancelled"}, MaxPending: 1, MaxAttempts: 3})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
task, err := api.SubmitTask(ctx, "task-"+id, orbit.SubmitTaskRequest{QueueID: queue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err = api.CancelTask(ctx, task.Task.ID, "cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "receiver process fault"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
first := startReceiverChild(t, s, created.Subscription.ID, "hold_after_commit")
|
||
committed := first.marker(t, ctx, "committed")
|
||
if committed.Duplicate {
|
||
t.Fatal("first receiver unexpectedly deduplicated")
|
||
}
|
||
view, err := s.View(ctx, task.Task.ID)
|
||
if err != nil || view.AppliedEvents != 1 || inboxCount(t, ctx, s) != 1 {
|
||
t.Fatal("receiver barrier occurred before durable commit", err)
|
||
}
|
||
before, err := api.ListEventDeliveries(ctx, created.Subscription.ID, orbit.DeliveryPageRequest{})
|
||
if err != nil || len(before.Items) != 1 || before.Items[0].State != orbit.DeliveryLeased || before.Items[0].AcknowledgedAt != nil {
|
||
t.Fatal("fault boundary is not before ACK", err)
|
||
}
|
||
first.kill(t)
|
||
second := startReceiverChild(t, s, created.Subscription.ID, "confirm_duplicate")
|
||
duplicate := second.marker(t, ctx, "committed")
|
||
ack := second.marker(t, ctx, "acknowledged")
|
||
if !duplicate.Duplicate || duplicate.EventID != committed.EventID || ack.EventID != committed.EventID {
|
||
t.Fatal("new process did not reconcile original event")
|
||
}
|
||
select {
|
||
case <-second.done:
|
||
if second.waitErr != nil {
|
||
t.Fatal("replacement receiver failed", second.waitErr)
|
||
}
|
||
case <-ctx.Done():
|
||
t.Fatal("replacement receiver did not exit cleanly")
|
||
}
|
||
after, err := s.View(ctx, task.Task.ID)
|
||
if err != nil || after.AppliedEvents != 1 || inboxCount(t, ctx, s) != 1 {
|
||
t.Fatal("business effect repeated after consumer restart", err)
|
||
}
|
||
confirmed, err := api.ListEventDeliveries(ctx, created.Subscription.ID, orbit.DeliveryPageRequest{State: orbit.DeliveryAcked})
|
||
if err != nil || len(confirmed.Items) != 1 || confirmed.Items[0].Attempts != 2 || confirmed.Items[0].ID != before.Items[0].ID {
|
||
t.Fatal("delivery recovery or acknowledgement history wrong", err)
|
||
}
|
||
t.Log("business commit verified before ACK; consumer SIGKILL confirmed, new process used durable Inbox after real lease expiry; one business effect and two delivery attempts")
|
||
}
|