89 lines
2.9 KiB
Go
89 lines
2.9 KiB
Go
// 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
|
|
}
|