From 5db48e1bccdfabaa78edf9ef5bdb5a4f425450b7 Mon Sep 17 00:00:00 2001 From: zhx Date: Mon, 13 Jul 2026 23:38:32 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=88=86=E5=8C=BA=E9=99=90?= =?UTF-8?q?=E9=87=8Foutbox=E5=9B=9E=E6=94=BE=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/outboxreplay/replay.go | 88 +++++++++ pkg/outboxreplay/replay_test.go | 81 +++++++++ services/game-service/Dockerfile | 3 +- .../game-service/cmd/replay-outbox/main.go | 59 ++++++ .../internal/app/outbox_replay.go | 63 +++++++ services/user-service/Dockerfile | 3 +- .../user-service/cmd/replay-outbox/main.go | 59 ++++++ .../internal/app/outbox_replay.go | 60 +++++++ services/wallet-service/Dockerfile | 3 +- .../wallet-service/cmd/replay-outbox/main.go | 79 ++++++++ .../cmd/replay-outbox/main_test.go | 19 ++ .../internal/app/outbox_replay.go | 168 ++++++++++++++++++ .../internal/app/outbox_replay_test.go | 43 +++++ 13 files changed, 725 insertions(+), 3 deletions(-) create mode 100644 pkg/outboxreplay/replay.go create mode 100644 pkg/outboxreplay/replay_test.go create mode 100644 services/game-service/cmd/replay-outbox/main.go create mode 100644 services/game-service/internal/app/outbox_replay.go create mode 100644 services/user-service/cmd/replay-outbox/main.go create mode 100644 services/user-service/internal/app/outbox_replay.go create mode 100644 services/wallet-service/cmd/replay-outbox/main.go create mode 100644 services/wallet-service/cmd/replay-outbox/main_test.go create mode 100644 services/wallet-service/internal/app/outbox_replay.go create mode 100644 services/wallet-service/internal/app/outbox_replay_test.go diff --git a/pkg/outboxreplay/replay.go b/pkg/outboxreplay/replay.go new file mode 100644 index 00000000..df8cfb14 --- /dev/null +++ b/pkg/outboxreplay/replay.go @@ -0,0 +1,88 @@ +// Package outboxreplay provides the process-level hard bound used by manual +// owner-outbox repairs. Normal workers intentionally keep polling; an operator +// replay must instead stop after an exact maximum even while the partition +// remains non-empty. +package outboxreplay + +import ( + "context" + "errors" + "fmt" + "strings" + + "hyapp/pkg/appcode" +) + +const ExecuteConfirmation = "REPLAY_EXACT_APP_OUTBOX_BATCH" + +// Scope is the immutable tenant and total-record bound for one process run. +// Limit is not a worker batch size: Run enforces it across every claim/publish +// iteration and returns once that many rows have been handled. +type Scope struct { + AppCode string + Limit int +} + +// Result reports whether the selected App partition became empty before the +// hard limit. Processed counts rows claimed by processOne, including a row that +// was released as retryable after a publish failure. +type Result struct { + Processed int + Drained bool +} + +// NewScope rejects an implicit tenant or an unbounded run before any database +// or MQ client is opened. maxLimit is command-specific so a service may impose +// a smaller operational envelope without weakening this shared contract. +func NewScope(rawAppCode string, limit int, maxLimit int) (Scope, error) { + if strings.TrimSpace(rawAppCode) == "" { + // appcode.Normalize intentionally preserves a legacy empty=>lalu default + // for request compatibility. An operator command must reject omission + // before normalization or a missing flag would target a real tenant. + return Scope{}, errors.New("app-code is required") + } + appCode := appcode.Normalize(rawAppCode) + if maxLimit <= 0 { + return Scope{}, errors.New("max limit must be positive") + } + if limit <= 0 || limit > maxLimit { + return Scope{}, fmt.Errorf("limit must be between 1 and %d", maxLimit) + } + return Scope{AppCode: appCode, Limit: limit}, nil +} + +// Run calls processOne at most Scope.Limit times. processOne must use the +// owner's ordinary indexed claim path with an effective batch size of one; +// accepting a larger count here would let an implementation silently strand +// already-claimed tail rows after this process reaches its advertised limit. +func Run(ctx context.Context, scope Scope, processOne func(context.Context) (int, error)) (Result, error) { + if ctx == nil { + ctx = context.Background() + } + if strings.TrimSpace(scope.AppCode) == "" || scope.Limit <= 0 { + return Result{}, errors.New("valid replay scope is required") + } + if processOne == nil { + return Result{}, errors.New("processOne is required") + } + + result := Result{} + for result.Processed < scope.Limit { + if err := ctx.Err(); err != nil { + return result, err + } + processed, err := processOne(ctx) + if processed < 0 || processed > 1 { + return result, fmt.Errorf("single-record replay processed %d rows", processed) + } + result.Processed += processed + if err != nil { + return result, err + } + if processed == 0 { + result.Drained = true + return result, nil + } + } + return result, nil +} diff --git a/pkg/outboxreplay/replay_test.go b/pkg/outboxreplay/replay_test.go new file mode 100644 index 00000000..28398ae2 --- /dev/null +++ b/pkg/outboxreplay/replay_test.go @@ -0,0 +1,81 @@ +package outboxreplay + +import ( + "context" + "errors" + "testing" +) + +func TestNewScopeNormalizesAndBoundsTenantReplay(t *testing.T) { + scope, err := NewScope(" HUWAA ", 3, 10) + if err != nil { + t.Fatalf("NewScope failed: %v", err) + } + if scope.AppCode != "huwaa" || scope.Limit != 3 { + t.Fatalf("scope mismatch: %+v", scope) + } + for _, test := range []struct { + name string + app string + limit int + max int + }{ + {name: "missing app", limit: 1, max: 10}, + {name: "zero limit", app: "huwaa", max: 10}, + {name: "above maximum", app: "huwaa", limit: 11, max: 10}, + {name: "invalid maximum", app: "huwaa", limit: 1}, + } { + t.Run(test.name, func(t *testing.T) { + if _, err := NewScope(test.app, test.limit, test.max); err == nil { + t.Fatal("expected validation error") + } + }) + } +} + +func TestRunStopsAtProcessWideLimit(t *testing.T) { + calls := 0 + result, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 3}, func(context.Context) (int, error) { + calls++ + return 1, nil + }) + if err != nil { + t.Fatalf("Run failed: %v", err) + } + if calls != 3 || result.Processed != 3 || result.Drained { + t.Fatalf("hard-limit result mismatch: calls=%d result=%+v", calls, result) + } +} + +func TestRunStopsWhenPartitionDrains(t *testing.T) { + calls := 0 + result, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 5}, func(context.Context) (int, error) { + calls++ + if calls == 3 { + return 0, nil + } + return 1, nil + }) + if err != nil { + t.Fatalf("Run failed: %v", err) + } + if calls != 3 || result.Processed != 2 || !result.Drained { + t.Fatalf("drained result mismatch: calls=%d result=%+v", calls, result) + } +} + +func TestRunCountsClaimedFailureAndRejectsBatchClaims(t *testing.T) { + publishErr := errors.New("publish failed") + result, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 5}, func(context.Context) (int, error) { + return 1, publishErr + }) + if !errors.Is(err, publishErr) || result.Processed != 1 { + t.Fatalf("failed row result mismatch: result=%+v err=%v", result, err) + } + + if _, err := Run(context.Background(), Scope{AppCode: "huwaa", Limit: 5}, func(context.Context) (int, error) { + return 2, nil + }); err == nil { + t.Fatal("expected multi-row claim rejection") + } +} diff --git a/services/game-service/Dockerfile b/services/game-service/Dockerfile index ad976db5..8cf511e8 100644 --- a/services/game-service/Dockerfile +++ b/services/game-service/Dockerfile @@ -15,13 +15,14 @@ COPY . . RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \ --mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \ mkdir -p /out && \ - GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/game-service/cmd/server ./cmd/grpc-health-probe + GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/game-service/cmd/server ./services/game-service/cmd/replay-outbox ./cmd/grpc-health-probe FROM alpine:3.20 ENV TZ=UTC WORKDIR /app RUN apk add --no-cache ca-certificates && adduser -D -g '' appuser COPY --from=builder /out/server /app/server +COPY --from=builder /out/replay-outbox /app/replay-outbox COPY --from=builder /out/grpc-health-probe /app/grpc-health-probe COPY services/game-service/configs/config.docker.yaml /app/config.yaml USER appuser diff --git a/services/game-service/cmd/replay-outbox/main.go b/services/game-service/cmd/replay-outbox/main.go new file mode 100644 index 00000000..93795ec8 --- /dev/null +++ b/services/game-service/cmd/replay-outbox/main.go @@ -0,0 +1,59 @@ +// Command replay-outbox publishes a hard-bounded game owner-outbox batch for +// one App without starting game APIs or downstream gRPC connections. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "os" + "strings" + "time" + + "hyapp/pkg/outboxreplay" + "hyapp/services/game-service/internal/app" + "hyapp/services/game-service/internal/config" +) + +func main() { + configPath := flag.String("config", "services/game-service/configs/config.yaml", "path to game-service config") + appCode := flag.String("app-code", "", "exact App partition to replay") + limit := flag.Int("limit", 0, "process-wide maximum rows, 1..500") + execute := flag.Bool("execute", false, "publish the bounded batch; default is a no-write dry-run") + confirmation := flag.String("confirm", "", "required execution confirmation") + timeout := flag.Duration("timeout", 2*time.Minute, "total command timeout, at most 30m") + flag.Parse() + + scope, err := outboxreplay.NewScope(*appCode, *limit, 500) + if err != nil { + fatal(err) + } + if *timeout <= 0 || *timeout > 30*time.Minute { + fatal(errors.New("timeout must be within (0,30m]")) + } + if *execute && strings.TrimSpace(*confirmation) != outboxreplay.ExecuteConfirmation { + fatal(fmt.Errorf("--execute requires --confirm=%s", outboxreplay.ExecuteConfirmation)) + } + cfg, err := config.Load(strings.TrimSpace(*configPath)) + if err != nil { + fatal(err) + } + if !*execute { + fmt.Printf("mode=dry-run app_code=%s limit=%d no_writes=true\n", scope.AppCode, scope.Limit) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + result, err := app.ReplayGameOutboxOnce(ctx, cfg, scope) + if err != nil { + fatal(fmt.Errorf("game replay stopped after %d rows: %w", result.Processed, err)) + } + fmt.Printf("mode=execute app_code=%s limit=%d processed=%d drained=%t\n", scope.AppCode, scope.Limit, result.Processed, result.Drained) +} + +func fatal(err error) { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) +} diff --git a/services/game-service/internal/app/outbox_replay.go b/services/game-service/internal/app/outbox_replay.go new file mode 100644 index 00000000..1dda7731 --- /dev/null +++ b/services/game-service/internal/app/outbox_replay.go @@ -0,0 +1,63 @@ +package app + +import ( + "context" + "errors" + "fmt" + "time" + + "hyapp/pkg/outboxpartition" + "hyapp/pkg/outboxreplay" + "hyapp/pkg/rocketmqx" + servicemq "hyapp/pkg/servicekit/mq" + "hyapp/services/game-service/internal/config" + mysqlstorage "hyapp/services/game-service/internal/storage/mysql" +) + +// ReplayGameOutboxOnce runs the ordinary game fact publisher for one App and a +// process-wide hard limit. It opens no gRPC listener and no downstream service +// connection, so an empty partition is a read-only no-op apart from MQ producer +// registration. +func ReplayGameOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope) (outboxreplay.Result, error) { + validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500) + if err != nil { + return outboxreplay.Result{}, err + } + if !cfg.RocketMQ.GameOutbox.Enabled { + return outboxreplay.Result{}, errors.New("game RocketMQ outbox is not enabled") + } + repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN) + if err != nil { + return outboxreplay.Result{}, err + } + defer func() { _ = repository.Close() }() + producer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ)) + if err != nil { + return outboxreplay.Result{}, err + } + if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil { + return outboxreplay.Result{}, err + } + defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer}) + + replayConfig := cfg + replayConfig.OutboxWorker.BatchSize = 1 + replayConfig.OutboxWorker.AppCodes = []string{validatedScope.AppCode} + replayCtx, cancel := context.WithCancel(ctx) + defer cancel() + replayApp := &App{ + repo: repository, + outboxProducer: producer, + outboxPartitions: outboxpartition.New(replayConfig.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval), + outboxCtx: replayCtx, + outboxCancel: cancel, + cfg: replayConfig, + } + workerID := fmt.Sprintf("ops-game-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli()) + return outboxreplay.Run(ctx, validatedScope, func(context.Context) (int, error) { + // processOutboxBatch derives the same worker prefix as the live service; + // attach the unique run ID to NodeID so lock ownership remains auditable. + replayApp.cfg.NodeID = workerID + return replayApp.processOutboxBatch() + }) +} diff --git a/services/user-service/Dockerfile b/services/user-service/Dockerfile index 937f8661..233771ea 100644 --- a/services/user-service/Dockerfile +++ b/services/user-service/Dockerfile @@ -19,7 +19,7 @@ COPY . . RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \ --mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \ mkdir -p /out && \ - GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/user-service/cmd/server ./cmd/grpc-health-probe + GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/user-service/cmd/server ./services/user-service/cmd/replay-outbox ./cmd/grpc-health-probe FROM alpine:3.20 @@ -30,6 +30,7 @@ WORKDIR /app RUN apk add --no-cache ca-certificates && adduser -D -g '' appuser COPY --from=builder /out/server /app/server +COPY --from=builder /out/replay-outbox /app/replay-outbox COPY --from=builder /out/grpc-health-probe /app/grpc-health-probe COPY services/user-service/configs/config.docker.yaml /app/config.yaml diff --git a/services/user-service/cmd/replay-outbox/main.go b/services/user-service/cmd/replay-outbox/main.go new file mode 100644 index 00000000..667f68cf --- /dev/null +++ b/services/user-service/cmd/replay-outbox/main.go @@ -0,0 +1,59 @@ +// Command replay-outbox publishes a hard-bounded user owner-outbox batch for +// one App without starting the user-service API or unrelated MQ consumers. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "os" + "strings" + "time" + + "hyapp/pkg/outboxreplay" + "hyapp/services/user-service/internal/app" + "hyapp/services/user-service/internal/config" +) + +func main() { + configPath := flag.String("config", "services/user-service/configs/config.yaml", "path to user-service config") + appCode := flag.String("app-code", "", "exact App partition to replay") + limit := flag.Int("limit", 0, "process-wide maximum rows, 1..500") + execute := flag.Bool("execute", false, "publish the bounded batch; default is a no-write dry-run") + confirmation := flag.String("confirm", "", "required execution confirmation") + timeout := flag.Duration("timeout", 2*time.Minute, "total command timeout, at most 30m") + flag.Parse() + + scope, err := outboxreplay.NewScope(*appCode, *limit, 500) + if err != nil { + fatal(err) + } + if *timeout <= 0 || *timeout > 30*time.Minute { + fatal(errors.New("timeout must be within (0,30m]")) + } + if *execute && strings.TrimSpace(*confirmation) != outboxreplay.ExecuteConfirmation { + fatal(fmt.Errorf("--execute requires --confirm=%s", outboxreplay.ExecuteConfirmation)) + } + cfg, err := config.Load(strings.TrimSpace(*configPath)) + if err != nil { + fatal(err) + } + if !*execute { + fmt.Printf("mode=dry-run app_code=%s limit=%d no_writes=true\n", scope.AppCode, scope.Limit) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + result, err := app.ReplayUserOutboxOnce(ctx, cfg, scope) + if err != nil { + fatal(fmt.Errorf("user replay stopped after %d rows: %w", result.Processed, err)) + } + fmt.Printf("mode=execute app_code=%s limit=%d processed=%d drained=%t\n", scope.AppCode, scope.Limit, result.Processed, result.Drained) +} + +func fatal(err error) { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) +} diff --git a/services/user-service/internal/app/outbox_replay.go b/services/user-service/internal/app/outbox_replay.go new file mode 100644 index 00000000..11efadd6 --- /dev/null +++ b/services/user-service/internal/app/outbox_replay.go @@ -0,0 +1,60 @@ +package app + +import ( + "context" + "errors" + "fmt" + "time" + + "hyapp/pkg/outboxpartition" + "hyapp/pkg/outboxreplay" + "hyapp/pkg/rocketmqx" + serviceapp "hyapp/pkg/servicekit/app" + servicemq "hyapp/pkg/servicekit/mq" + "hyapp/services/user-service/internal/config" + mysqlstorage "hyapp/services/user-service/internal/storage/mysql" +) + +// ReplayUserOutboxOnce processes a process-wide bounded App batch without +// starting user-service listeners or its unrelated consumers. Claiming one row +// per iteration keeps the existing 30-second row lease attached only to the +// message currently being published. +func ReplayUserOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope) (outboxreplay.Result, error) { + validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500) + if err != nil { + return outboxreplay.Result{}, err + } + if !cfg.RocketMQ.UserOutbox.Enabled { + return outboxreplay.Result{}, errors.New("user RocketMQ outbox is not enabled") + } + repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN) + if err != nil { + return outboxreplay.Result{}, err + } + defer func() { _ = repository.Close() }() + producer, err := rocketmqx.NewProducer(userOutboxProducerConfig(cfg.RocketMQ)) + if err != nil { + return outboxreplay.Result{}, err + } + if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil { + return outboxreplay.Result{}, err + } + defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer}) + + replayConfig := cfg + replayConfig.OutboxWorker.BatchSize = 1 + replayConfig.OutboxWorker.AppCodes = []string{validatedScope.AppCode} + workers := serviceapp.NewBackground(ctx) + defer workers.StopAndWait() + replayApp := &App{ + mysqlRepo: repository, + userOutboxProducer: producer, + userOutboxPartitions: outboxpartition.New(replayConfig.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval), + cfg: replayConfig, + workers: workers, + } + workerID := fmt.Sprintf("ops-user-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli()) + return outboxreplay.Run(ctx, validatedScope, func(context.Context) (int, error) { + return replayApp.processUserOutboxBatch(workerID) + }) +} diff --git a/services/wallet-service/Dockerfile b/services/wallet-service/Dockerfile index ea0b1fac..71567241 100644 --- a/services/wallet-service/Dockerfile +++ b/services/wallet-service/Dockerfile @@ -19,7 +19,7 @@ COPY . . RUN --mount=type=cache,id=hyapp-go-mod,target=/go/pkg/mod \ --mount=type=cache,id=hyapp-go-build,target=/root/.cache/go-build \ mkdir -p /out && \ - GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/wallet-service/cmd/server ./cmd/grpc-health-probe + GOWORK=off CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -o /out/ ./services/wallet-service/cmd/server ./services/wallet-service/cmd/replay-outbox ./cmd/grpc-health-probe FROM alpine:3.20 @@ -30,6 +30,7 @@ WORKDIR /app RUN apk add --no-cache ca-certificates && adduser -D -g '' appuser COPY --from=builder /out/server /app/server +COPY --from=builder /out/replay-outbox /app/replay-outbox COPY --from=builder /out/grpc-health-probe /app/grpc-health-probe COPY services/wallet-service/configs/config.docker.yaml /app/config.yaml diff --git a/services/wallet-service/cmd/replay-outbox/main.go b/services/wallet-service/cmd/replay-outbox/main.go new file mode 100644 index 00000000..15c2356e --- /dev/null +++ b/services/wallet-service/cmd/replay-outbox/main.go @@ -0,0 +1,79 @@ +// Command replay-outbox publishes one explicitly bounded wallet owner-outbox +// batch. It is intentionally separate from the long-running server worker: +// batch-size there controls one poll, while --limit here caps the whole process. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "os" + "strings" + "time" + + "hyapp/pkg/outboxreplay" + "hyapp/services/wallet-service/internal/app" + "hyapp/services/wallet-service/internal/config" +) + +type repeatedFlag []string + +func (values *repeatedFlag) String() string { return strings.Join(*values, ",") } + +func (values *repeatedFlag) Set(value string) error { + value = strings.TrimSpace(value) + if value == "" { + return errors.New("event-type cannot be empty") + } + *values = append(*values, value) + return nil +} + +func main() { + configPath := flag.String("config", "services/wallet-service/configs/config.yaml", "path to wallet-service config") + appCode := flag.String("app-code", "", "exact App partition to replay") + limit := flag.Int("limit", 0, "process-wide maximum rows, 1..500") + execute := flag.Bool("execute", false, "publish the bounded batch; default is a no-write dry-run") + confirmation := flag.String("confirm", "", "required execution confirmation") + timeout := flag.Duration("timeout", 2*time.Minute, "total command timeout, at most 30m") + var eventTypes repeatedFlag + flag.Var(&eventTypes, "event-type", "exact wallet event type to include; repeat for multiple types") + flag.Parse() + + scope, err := outboxreplay.NewScope(*appCode, *limit, 500) + if err != nil { + fatal(err) + } + if *timeout <= 0 || *timeout > 30*time.Minute { + fatal(errors.New("timeout must be within (0,30m]")) + } + if *execute && strings.TrimSpace(*confirmation) != outboxreplay.ExecuteConfirmation { + fatal(fmt.Errorf("--execute requires --confirm=%s", outboxreplay.ExecuteConfirmation)) + } + cfg, err := config.Load(strings.TrimSpace(*configPath)) + if err != nil { + fatal(err) + } + normalizedTypes, err := app.ValidateWalletReplayEventTypes(cfg, eventTypes) + if err != nil { + fatal(err) + } + if !*execute { + fmt.Printf("mode=dry-run app_code=%s limit=%d event_types=%s no_writes=true\n", scope.AppCode, scope.Limit, strings.Join(normalizedTypes, ",")) + return + } + + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + result, err := app.ReplayWalletOutboxOnce(ctx, cfg, scope, normalizedTypes) + if err != nil { + fatal(fmt.Errorf("wallet replay stopped after %d rows: %w", result.Processed, err)) + } + fmt.Printf("mode=execute app_code=%s limit=%d processed=%d drained=%t event_types=%s\n", scope.AppCode, scope.Limit, result.Processed, result.Drained, strings.Join(normalizedTypes, ",")) +} + +func fatal(err error) { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) +} diff --git a/services/wallet-service/cmd/replay-outbox/main_test.go b/services/wallet-service/cmd/replay-outbox/main_test.go new file mode 100644 index 00000000..da76a324 --- /dev/null +++ b/services/wallet-service/cmd/replay-outbox/main_test.go @@ -0,0 +1,19 @@ +package main + +import "testing" + +func TestRepeatedEventTypeFlagAcceptsMultipleAndRejectsEmpty(t *testing.T) { + var values repeatedFlag + if err := values.Set(" GiftConfigChanged "); err != nil { + t.Fatalf("first event type failed: %v", err) + } + if err := values.Set("ResourceChanged"); err != nil { + t.Fatalf("second event type failed: %v", err) + } + if len(values) != 2 || values[0] != "GiftConfigChanged" || values[1] != "ResourceChanged" { + t.Fatalf("repeated event types mismatch: %v", values) + } + if err := values.Set(" "); err == nil { + t.Fatal("expected empty event type rejection") + } +} diff --git a/services/wallet-service/internal/app/outbox_replay.go b/services/wallet-service/internal/app/outbox_replay.go new file mode 100644 index 00000000..bb96875d --- /dev/null +++ b/services/wallet-service/internal/app/outbox_replay.go @@ -0,0 +1,168 @@ +package app + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "hyapp/pkg/appcode" + "hyapp/pkg/outboxreplay" + "hyapp/pkg/rocketmqx" + servicemq "hyapp/pkg/servicekit/mq" + "hyapp/pkg/walletmq" + "hyapp/services/wallet-service/internal/config" + mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql" +) + +// ReplayWalletOutboxOnce publishes at most scope.Limit claimable facts from +// exactly one App partition, then closes both MySQL and MQ. Each iteration uses +// the production claim/publish/mark path with batch size one, so a process +// timeout cannot leave a claimed tail waiting for lease expiry. +func ReplayWalletOutboxOnce(ctx context.Context, cfg config.Config, scope outboxreplay.Scope, eventTypes []string) (outboxreplay.Result, error) { + validatedScope, err := outboxreplay.NewScope(scope.AppCode, scope.Limit, 500) + if err != nil { + return outboxreplay.Result{}, err + } + includeEventTypes, err := ValidateWalletReplayEventTypes(cfg, eventTypes) + if err != nil { + return outboxreplay.Result{}, err + } + if !cfg.RocketMQ.WalletOutbox.Enabled { + return outboxreplay.Result{}, errors.New("wallet RocketMQ outbox is not enabled") + } + + repository, err := mysqlstorage.Open(ctx, cfg.MySQLDSN) + if err != nil { + return outboxreplay.Result{}, err + } + defer func() { _ = repository.Close() }() + producer, err := rocketmqx.NewProducer(rocketMQProducerConfig(cfg.RocketMQ, cfg.RocketMQ.WalletOutbox.ProducerGroup)) + if err != nil { + return outboxreplay.Result{}, err + } + if err := servicemq.StartProducers([]*rocketmqx.Producer{producer}); err != nil { + return outboxreplay.Result{}, err + } + defer servicemq.ShutdownProducers([]*rocketmqx.Producer{producer}) + + workerOptions := cfg.OutboxWorker + workerOptions.BatchSize = 1 + workerOptions.Concurrency = 1 + replayApp := &App{ + mysqlRepo: repository, + walletEventTagMode: cfg.RocketMQ.WalletEventTagMode, + walletOutboxTopic: cfg.RocketMQ.WalletOutbox.Topic, + outboxWorkerCfg: workerOptions, + } + workerID := fmt.Sprintf("ops-wallet-outbox-%s-%d", validatedScope.AppCode, time.Now().UTC().UnixMilli()) + return outboxreplay.Run(ctx, validatedScope, func(runCtx context.Context) (int, error) { + return replayApp.replayOneWalletOutboxRecord(runCtx, workerID, workerOptions, producer, cfg.RocketMQ.WalletOutbox.Topic, validatedScope.AppCode, includeEventTypes) + }) +} + +func (a *App) replayOneWalletOutboxRecord(ctx context.Context, workerID string, options config.OutboxWorkerConfig, producer *rocketmqx.Producer, topic string, targetAppCode string, includeEventTypes []string) (int, error) { + if options.PublishTimeout <= 0 { + options.PublishTimeout = 3 * time.Second + } + claimCtx := appcode.WithContext(ctx, targetAppCode) + records, err := a.mysqlRepo.ClaimPendingWalletOutboxFiltered( + claimCtx, + workerID, + 1, + time.Now().UTC().Add(walletOutboxSingleRecordLease(options.PublishTimeout)).UnixMilli(), + mysqlstorage.WalletOutboxClaimFilter{IncludeEventTypes: includeEventTypes}, + ) + if err != nil || len(records) == 0 { + return 0, err + } + record := records[0] + if record.RetryCount >= options.MaxRetryCount { + markCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) + markErr := a.mysqlRepo.MarkWalletOutboxDead(markCtx, record.EventID, deadWalletOutboxReason(record)) + cancel() + if markErr != nil { + return 1, markErr + } + return 1, fmt.Errorf("wallet outbox event %s reached max retry count and was marked dead", record.EventID) + } + + publishCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) + publishErr := a.publishWalletOutboxRecord(publishCtx, producer, topic, record) + cancel() + markCtx, markCancel := context.WithTimeout(appcode.WithContext(context.Background(), record.AppCode), options.PublishTimeout) + defer markCancel() + if publishErr != nil { + nextRetryAtMS := time.Now().UTC().Add(walletOutboxBackoff(record.RetryCount+1, options)).UnixMilli() + if markErr := a.mysqlRepo.MarkWalletOutboxRetryable(markCtx, record.EventID, publishErr.Error(), nextRetryAtMS); markErr != nil { + return 1, errors.Join(publishErr, markErr) + } + // The row was safely released for a later retry, but this explicit run must + // fail instead of reporting a merely claimed row as a delivered replay. + return 1, publishErr + } + if err := a.mysqlRepo.MarkWalletOutboxDelivered(markCtx, record.EventID); err != nil { + return 1, err + } + return 1, nil +} + +// NormalizeWalletReplayEventTypes validates the exact MQ tags before an +// operator dry-run is promoted to an executing run. +func NormalizeWalletReplayEventTypes(values []string) ([]string, error) { + if len(values) > 32 { + return nil, errors.New("wallet event-type count cannot exceed 32") + } + normalized := make([]string, 0, len(values)) + seen := make(map[string]struct{}, len(values)) + for _, value := range values { + eventType, err := walletmq.EventTypeTag(strings.TrimSpace(value)) + if err != nil { + return nil, err + } + if _, exists := seen[eventType]; exists { + continue + } + seen[eventType] = struct{}{} + normalized = append(normalized, eventType) + } + return normalized, nil +} + +// ValidateWalletReplayEventTypes applies the lane boundary in both dry-run and +// execute modes. The normal-topic repair must reject realtime event types even +// while that worker or topic is temporarily disabled; pausing a lane does not +// transfer ownership of its durable rows to another topic. +func ValidateWalletReplayEventTypes(cfg config.Config, values []string) ([]string, error) { + normalized, err := NormalizeWalletReplayEventTypes(values) + if err != nil { + return nil, err + } + if len(normalized) == 0 { + // Wallet fanout has both high-volume balance notices and low-volume config + // facts. Requiring an explicit set prevents a small operational command + // from accidentally mixing stages merely because their timestamps overlap. + return nil, errors.New("at least one wallet event-type is required") + } + if err := rejectRealtimeReplayEventTypes(normalized, cfg.RealtimeOutboxWorker.EventTypes); err != nil { + return nil, err + } + return normalized, nil +} + +func rejectRealtimeReplayEventTypes(included []string, realtime []string) error { + realtimeSet := make(map[string]struct{}, len(realtime)) + for _, eventType := range realtime { + eventType = strings.TrimSpace(eventType) + if eventType != "" { + realtimeSet[eventType] = struct{}{} + } + } + for _, eventType := range included { + if _, exists := realtimeSet[eventType]; exists { + return fmt.Errorf("wallet event-type %s belongs to the realtime outbox lane", eventType) + } + } + return nil +} diff --git a/services/wallet-service/internal/app/outbox_replay_test.go b/services/wallet-service/internal/app/outbox_replay_test.go new file mode 100644 index 00000000..41d2a6fe --- /dev/null +++ b/services/wallet-service/internal/app/outbox_replay_test.go @@ -0,0 +1,43 @@ +package app + +import ( + "context" + "strings" + "testing" + + "hyapp/pkg/outboxreplay" + "hyapp/services/wallet-service/internal/config" +) + +func TestNormalizeWalletReplayEventTypes(t *testing.T) { + eventTypes, err := NormalizeWalletReplayEventTypes([]string{" GiftConfigChanged ", "WalletBalanceChanged", "GiftConfigChanged"}) + if err != nil { + t.Fatalf("NormalizeWalletReplayEventTypes failed: %v", err) + } + if len(eventTypes) != 2 || eventTypes[0] != "GiftConfigChanged" || eventTypes[1] != "WalletBalanceChanged" { + t.Fatalf("normalized event types mismatch: %v", eventTypes) + } + if _, err := NormalizeWalletReplayEventTypes([]string{"bad || tag"}); err == nil { + t.Fatal("expected invalid RocketMQ tag rejection") + } +} + +func TestRejectRealtimeReplayEventTypes(t *testing.T) { + if err := rejectRealtimeReplayEventTypes([]string{"GiftConfigChanged"}, []string{"WalletRedPacketCreated"}); err != nil { + t.Fatalf("normal event rejected: %v", err) + } + if err := rejectRealtimeReplayEventTypes([]string{"WalletRedPacketCreated"}, []string{"WalletRedPacketCreated"}); err == nil { + t.Fatal("expected realtime lane rejection") + } +} + +func TestWalletReplayRejectsRealtimeTypeWhenLaneIsDisabled(t *testing.T) { + cfg := config.Default() + cfg.RealtimeOutboxWorker.Enabled = false + cfg.RocketMQ.RealtimeOutbox.Enabled = false + cfg.RealtimeOutboxWorker.EventTypes = []string{"WalletRedPacketCreated"} + _, err := ReplayWalletOutboxOnce(context.Background(), cfg, outboxreplay.Scope{AppCode: "huwaa", Limit: 1}, []string{"WalletRedPacketCreated"}) + if err == nil || !strings.Contains(err.Error(), "realtime outbox lane") { + t.Fatalf("disabled realtime lane must retain event ownership, err=%v", err) + } +}