feat: add durable business inbox example and consumer crash recovery

2026-09-17 20:49:51 +08:00
parent b17d9b1977
commit a8a8dcfe1e
27 changed files with 1601 additions and 17 deletions
--- a/examples/result-inbox/internal/inbox/consumer_process_integration_test.go
+++ b/examples/result-inbox/internal/inbox/consumer_process_integration_test.go
@@ -0,0 +1,298 @@
+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")
+}