MiFaZhan
475 lines
17 KiB
Go
475 lines
17 KiB
Go
package runtime
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gogf/gf/v2/container/gvar"
|
|
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/contract"
|
|
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
|
)
|
|
|
|
// SimpleReplay hands every source failure to a separate Run. The source hash
|
|
// is the checkpoint: a record is removed only after the target's task-state
|
|
// confirms that EnqueueRunBatch accepted it. No cursor is retained across
|
|
// mutations of the source hash.
|
|
//
|
|
// The target's retry-round budget is taken from the caller when supplied and
|
|
// falls back to the configured default otherwise. It is persisted on the
|
|
// source reservation because a worker restart re-enters through reconcile(),
|
|
// which no longer has the caller's arguments.
|
|
type SimpleReplay struct {
|
|
stores []*zset.Store
|
|
redis zset.RedisClient
|
|
// defaultMaxRounds applies when a replay request does not specify one.
|
|
defaultMaxRounds int
|
|
}
|
|
|
|
// Replay round bounds. MaxReplayRounds is a server-side guard, not a capacity
|
|
// limit: the observed chain of 10-09 needed 74 rounds to reach 98% recovery,
|
|
// so a low ceiling would cut off a legitimate convergence path.
|
|
//
|
|
// DefaultReplayMaxRounds is only the fallback for a request that omits a
|
|
// budget; callers may ask for more, up to MaxReplayRounds.
|
|
const (
|
|
DefaultReplayMaxRounds = 3
|
|
MaxReplayRounds = 100
|
|
legacyReplayMaxRounds = 1
|
|
)
|
|
|
|
type ReplayReceipt struct {
|
|
ReplayRunID string `json:"replayRunId"`
|
|
Status string `json:"status"`
|
|
}
|
|
|
|
type ReplayStatus struct {
|
|
ReplayRunID string `json:"replayRunId"`
|
|
SourceRunID string `json:"sourceRunId"`
|
|
Status string `json:"status"`
|
|
Remaining int64 `json:"remaining"`
|
|
Enqueued int64 `json:"enqueued"`
|
|
RunStatus string `json:"runStatus,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
}
|
|
|
|
var ErrNoFailures = errors.New("run has no failures to replay")
|
|
|
|
func NewSimpleReplay(stores ...*zset.Store) (*SimpleReplay, error) {
|
|
return NewSimpleReplayWithDefault(DefaultReplayMaxRounds, stores...)
|
|
}
|
|
|
|
// NewSimpleReplayWithDefault builds a replay service whose unspecified replay
|
|
// requests inherit defaultMaxRounds. A non-positive value falls back to
|
|
// DefaultReplayMaxRounds so a missing config key cannot disable the budget.
|
|
func NewSimpleReplayWithDefault(defaultMaxRounds int, stores ...*zset.Store) (*SimpleReplay, error) {
|
|
if len(stores) == 0 || stores[0] == nil {
|
|
return nil, fmt.Errorf("replay requires Redis stores")
|
|
}
|
|
for _, store := range stores {
|
|
if store == nil {
|
|
return nil, fmt.Errorf("replay store is nil")
|
|
}
|
|
}
|
|
if defaultMaxRounds <= 0 {
|
|
defaultMaxRounds = DefaultReplayMaxRounds
|
|
}
|
|
if defaultMaxRounds > MaxReplayRounds {
|
|
defaultMaxRounds = MaxReplayRounds
|
|
}
|
|
return &SimpleReplay{stores: stores, redis: stores[0].RedisClient(), defaultMaxRounds: defaultMaxRounds}, nil
|
|
}
|
|
|
|
func (r *SimpleReplay) find(ctx context.Context, id string) (*zset.Store, zset.RunMeta, error) {
|
|
if err := contract.ValidateID(id); err != nil || len(id) > 128 {
|
|
return nil, zset.RunMeta{}, contract.ErrInvalidArgument
|
|
}
|
|
var found *zset.Store
|
|
var meta zset.RunMeta
|
|
for _, store := range r.stores {
|
|
key, _ := store.RunMetaKey(id)
|
|
v, err := r.redis.Do(ctx, "EXISTS", key)
|
|
if err != nil {
|
|
return nil, meta, err
|
|
}
|
|
if v.Int() == 0 {
|
|
continue
|
|
}
|
|
if found != nil {
|
|
return nil, meta, contract.ErrConflict
|
|
}
|
|
found = store
|
|
meta, err = store.GetRunMeta(ctx, id)
|
|
if err != nil {
|
|
return nil, meta, err
|
|
}
|
|
}
|
|
if found == nil {
|
|
return nil, meta, contract.ErrNotFound
|
|
}
|
|
return found, meta, nil
|
|
}
|
|
|
|
// Request accepts a replay of the source run's failures. maxRounds == 0 means
|
|
// "use the configured default"; requests above MaxReplayRounds are rejected.
|
|
func (r *SimpleReplay) Request(ctx context.Context, sourceID, requestedBy string, maxRounds int) (ReplayReceipt, error) {
|
|
if len(requestedBy) == 0 || len(requestedBy) > 64 {
|
|
return ReplayReceipt{}, contract.ErrInvalidArgument
|
|
}
|
|
if maxRounds < 0 || maxRounds > MaxReplayRounds {
|
|
return ReplayReceipt{}, contract.ErrInvalidArgument
|
|
}
|
|
if maxRounds == 0 {
|
|
maxRounds = r.defaultMaxRounds
|
|
}
|
|
store, _, err := r.find(ctx, sourceID)
|
|
if err != nil {
|
|
return ReplayReceipt{}, err
|
|
}
|
|
metaKey, _ := store.RunMetaKey(sourceID)
|
|
failedKey, _ := store.FailedKey(sourceID)
|
|
var random [16]byte
|
|
if _, err := rand.Read(random[:]); err != nil {
|
|
return ReplayReceipt{}, err
|
|
}
|
|
targetID := "replay-" + hex.EncodeToString(random[:])
|
|
// An index written before the reservation makes the pending source
|
|
// discoverable even if the process stops immediately after the Lua call.
|
|
if _, err = r.redis.Do(ctx, "SADD", store.Prefix()+"replay:pending", sourceID); err != nil {
|
|
return ReplayReceipt{}, err
|
|
}
|
|
v, err := r.redis.Eval(ctx, reserveReplay, 2, []string{metaKey, failedKey}, []any{targetID, requestedBy, maxRounds})
|
|
if err != nil {
|
|
return ReplayReceipt{}, err
|
|
}
|
|
result := v.String()
|
|
// An unsuccessful reservation leaves an inert discovery entry. Never
|
|
// remove it here: a concurrent caller might have reserved the source
|
|
// after this script returned, and that reservation needs recovery.
|
|
switch result {
|
|
case "missing":
|
|
return ReplayReceipt{}, contract.ErrNotFound
|
|
case "running":
|
|
return ReplayReceipt{}, contract.ErrConflict
|
|
case "empty":
|
|
return ReplayReceipt{}, ErrNoFailures
|
|
}
|
|
if !strings.HasPrefix(result, "replay-") {
|
|
return ReplayReceipt{}, fmt.Errorf("unexpected replay reservation result")
|
|
}
|
|
go r.process(store, sourceID)
|
|
return ReplayReceipt{ReplayRunID: result, Status: "accepted"}, nil
|
|
}
|
|
|
|
// Start reconciles pending sources after restarts and missed goroutine wakes.
|
|
// Every process may run it: the per-source Redis lease elects one writer.
|
|
func (r *SimpleReplay) Start(ctx context.Context) {
|
|
go func() {
|
|
ticker := time.NewTicker(10 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
r.reconcile(ctx)
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (r *SimpleReplay) reconcile(ctx context.Context) {
|
|
for _, store := range r.stores {
|
|
v, err := r.redis.Do(ctx, "SMEMBERS", store.Prefix()+"replay:pending")
|
|
if err != nil {
|
|
continue
|
|
}
|
|
for _, id := range v.Strings() {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
key, keyErr := store.RunMetaKey(id)
|
|
if keyErr != nil {
|
|
continue
|
|
}
|
|
reservation, getErr := r.redis.Do(ctx, "HGET", key, "replayTargetId")
|
|
// SADD happens before the reservation so even a crash during
|
|
// acceptance cannot lose work. Unreserved entries are inert.
|
|
if getErr != nil || reservation == nil || reservation.String() == "" {
|
|
continue
|
|
}
|
|
state, stateErr := r.redis.Do(ctx, "HGET", key, "replayState")
|
|
if stateErr != nil || state == nil || state.String() == "needs_attention" {
|
|
continue
|
|
}
|
|
go r.process(store, id)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *SimpleReplay) Status(ctx context.Context, targetID string) (ReplayStatus, error) {
|
|
store, target, err := r.find(ctx, targetID)
|
|
if err != nil {
|
|
// The target may not exist yet: locate its durable source reservation.
|
|
if !errors.Is(err, contract.ErrNotFound) {
|
|
return ReplayStatus{}, err
|
|
}
|
|
for _, candidate := range r.stores {
|
|
v, e := r.redis.Do(ctx, "SMEMBERS", candidate.Prefix()+"replay:pending")
|
|
if e != nil {
|
|
return ReplayStatus{}, e
|
|
}
|
|
for _, sourceID := range v.Strings() {
|
|
key, _ := candidate.RunMetaKey(sourceID)
|
|
value, e := r.redis.Do(ctx, "HGET", key, "replayTargetId")
|
|
if e == nil && value != nil && value.String() == targetID {
|
|
store = candidate
|
|
target.SourceRunID = sourceID
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if store == nil {
|
|
return ReplayStatus{}, contract.ErrNotFound
|
|
}
|
|
}
|
|
if target.SourceRunID == "" {
|
|
return ReplayStatus{}, contract.ErrNotFound
|
|
}
|
|
sourceKey, _ := store.RunMetaKey(target.SourceRunID)
|
|
failureKey, _ := store.FailedKey(target.SourceRunID)
|
|
values, err := r.redis.Do(ctx, "HMGET", sourceKey, "replayTargetId", "replayState", "replayError", "replayEnqueued")
|
|
if err != nil {
|
|
return ReplayStatus{}, err
|
|
}
|
|
fields := values.Vars()
|
|
if len(fields) != 4 || fields[0].String() != targetID {
|
|
return ReplayStatus{}, contract.ErrNotFound
|
|
}
|
|
remaining, err := r.redis.Do(ctx, "HLEN", failureKey)
|
|
if err != nil {
|
|
return ReplayStatus{}, err
|
|
}
|
|
return ReplayStatus{ReplayRunID: targetID, SourceRunID: target.SourceRunID, Status: fields[1].String(), Error: fields[2].String(), Enqueued: fields[3].Int64(), Remaining: remaining.Int64(), RunStatus: target.Status}, nil
|
|
}
|
|
|
|
func (r *SimpleReplay) process(store *zset.Store, sourceID string) {
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
|
|
defer cancel()
|
|
// Keep one process per source; batches are bounded so a short lease is
|
|
// renewed before it expires. A lost lease stops cleanup at the next batch.
|
|
lockKey := store.Prefix() + "{" + sourceID + "}:replay-lock"
|
|
var token [16]byte
|
|
if _, err := rand.Read(token[:]); err != nil {
|
|
return
|
|
}
|
|
owner := hex.EncodeToString(token[:])
|
|
locked, err := store.AcquireLock(ctx, lockKey, owner, time.Minute)
|
|
if err != nil || !locked {
|
|
return
|
|
}
|
|
defer func() { _, _ = store.ReleaseLock(context.Background(), lockKey, owner) }()
|
|
if err := r.handoff(ctx, store, sourceID, lockKey, owner); err != nil {
|
|
// The time slice ended, not the replay: the reconciler will resume
|
|
// from remaining source records under a new lease.
|
|
if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
|
|
return
|
|
}
|
|
sourceKey, _ := store.RunMetaKey(sourceID)
|
|
// A stale owner must not overwrite a newer worker's status.
|
|
_, _ = r.redis.Eval(context.Background(), replayAttention, 2, []string{sourceKey, lockKey}, []any{owner})
|
|
}
|
|
}
|
|
|
|
func (r *SimpleReplay) handoff(ctx context.Context, store *zset.Store, sourceID, lockKey, owner string) error {
|
|
sourceKey, _ := store.RunMetaKey(sourceID)
|
|
failedKey, _ := store.FailedKey(sourceID)
|
|
meta, err := store.GetRunMeta(ctx, sourceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
v, err := r.redis.Do(ctx, "HGET", sourceKey, "replayTargetId")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if v == nil || v.String() == "" {
|
|
return fmt.Errorf("replay reservation missing")
|
|
}
|
|
targetID := v.String()
|
|
// New reservations persist their budget. Reservations made before this
|
|
// field existed used one round; a restart must not apply today's default
|
|
// to those already accepted runs.
|
|
storedMaxRounds, err := r.redis.Do(ctx, "HGET", sourceKey, "replayMaxRound")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
maxRounds := legacyReplayMaxRounds
|
|
if storedMaxRounds != nil && storedMaxRounds.String() != "" {
|
|
maxRounds = storedMaxRounds.Int()
|
|
}
|
|
if maxRounds <= 0 {
|
|
return fmt.Errorf("replay reservation has invalid round budget")
|
|
}
|
|
if maxRounds > MaxReplayRounds {
|
|
return fmt.Errorf("replay reservation exceeds round budget limit")
|
|
}
|
|
// CreateRun is idempotent for the main hash but writes extension fields
|
|
// again. Read and validate an existing target BEFORE CreateRun, otherwise
|
|
// recovery could silently replace its accepted maxRound.
|
|
existing, err := store.GetRunMeta(ctx, targetID)
|
|
if err != nil && !errors.Is(err, zset.ErrRunNotFound) {
|
|
return err
|
|
}
|
|
if err == nil {
|
|
if storedMaxRounds == nil || storedMaxRounds.String() == "" {
|
|
maxRounds = existing.MaxRound
|
|
}
|
|
if maxRounds <= 0 || maxRounds > MaxReplayRounds || existing.SourceRunID != sourceID || existing.SyncType != "replay" || existing.MaxRound != maxRounds || existing.Operation != meta.Operation || existing.QueryMonth != meta.QueryMonth {
|
|
return fmt.Errorf("replay target metadata is inconsistent")
|
|
}
|
|
}
|
|
target := zset.RunMeta{RunID: targetID, Operation: meta.Operation, QueryMonth: meta.QueryMonth, SyncType: "replay", SourceRunID: sourceID, ReplayCycle: meta.ReplayCycle + 1, TriggerSource: "replay", MaxRound: maxRounds}
|
|
if err := store.CreateRun(ctx, target); err != nil {
|
|
return err
|
|
}
|
|
actual, err := store.GetRunMeta(ctx, targetID)
|
|
if err != nil || actual.SourceRunID != sourceID || actual.SyncType != "replay" || actual.MaxRound != maxRounds || actual.Operation != meta.Operation || actual.QueryMonth != meta.QueryMonth {
|
|
return fmt.Errorf("replay target metadata is inconsistent")
|
|
}
|
|
// The target task-state is the admission receipt during recovery. It may
|
|
// not expire while source failures are still being transferred.
|
|
targetKey, _ := store.RunMetaKey(targetID)
|
|
stateKey, _ := store.TaskStateKey(targetID)
|
|
if !actual.ProducerDone {
|
|
if _, err := r.redis.Do(ctx, "PERSIST", targetKey); err != nil {
|
|
return err
|
|
}
|
|
if _, err := r.redis.Do(ctx, "PERSIST", stateKey); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
_, err = r.redis.Do(ctx, "HSET", sourceKey, "replayState", "processing", "replayError", "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for {
|
|
if ctx.Err() != nil {
|
|
return ctx.Err()
|
|
}
|
|
valid, err := store.RenewLock(ctx, lockKey, owner, time.Minute)
|
|
if err != nil || !valid {
|
|
return fmt.Errorf("replay lease lost: %w", err)
|
|
}
|
|
// HSCAN is not a snapshot. Complete a bounded cursor pass, then
|
|
// recheck HLEN and repeat from zero if deletion changed the scan.
|
|
var cursor uint64
|
|
for {
|
|
v, e := r.redis.Do(ctx, "HSCAN", failedKey, cursor, "COUNT", 100)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
var scanPairs []*gvar.Var
|
|
cursor, scanPairs, e = parseScanResult(v)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if len(scanPairs) > 0 {
|
|
for i := 0; i < len(scanPairs); i += 2 {
|
|
valid, renewErr := store.RenewLock(ctx, lockKey, owner, time.Minute)
|
|
if renewErr != nil || !valid {
|
|
return fmt.Errorf("replay lease lost")
|
|
}
|
|
field, raw := scanPairs[i].String(), scanPairs[i+1].String()
|
|
var rec zset.FailureRecord
|
|
if json.Unmarshal([]byte(raw), &rec) != nil || rec.TaskID == "" || rec.CardID <= 0 || rec.ICCID == "" || rec.ChanKey == "" || rec.TempKey == "" || field != rec.TaskID || (rec.LatestRunID != "" && rec.LatestRunID != sourceID) {
|
|
return fmt.Errorf("source failure record cannot be replayed")
|
|
}
|
|
operation := rec.Operation
|
|
if operation == "" {
|
|
operation = meta.Operation
|
|
}
|
|
// Source task identity (not just card ID) prevents collisions
|
|
// between distinct failures for the same card.
|
|
taskID := targetID + ":" + rec.TaskID
|
|
queryMonth := rec.QueryMonth
|
|
if queryMonth == "" {
|
|
queryMonth = meta.QueryMonth
|
|
}
|
|
task := zset.Task{RunID: targetID, TaskID: taskID, Operation: operation, QueryMonth: queryMonth, CardID: rec.CardID, ICCID: rec.ICCID, MSISDN: rec.MSISDN, IMSI: rec.IMSI, ChanKey: rec.ChanKey, TempKey: rec.TempKey, PackageCode: rec.PackageCode}
|
|
if _, e = store.EnqueueRunBatch(ctx, targetID, rec.ChanKey, []zset.Task{task}, time.Now()); e != nil {
|
|
return e
|
|
}
|
|
confirmed, e := r.redis.Do(ctx, "HEXISTS", stateKey, taskID)
|
|
if e != nil || confirmed == nil || confirmed.Int() != 1 {
|
|
return fmt.Errorf("target enqueue unconfirmed: %w", e)
|
|
}
|
|
removed, e := r.redis.Eval(ctx, cleanupReplayFailure, 3, []string{sourceKey, failedKey, lockKey}, []any{field, raw, targetID, owner})
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if removed.Int() == -1 {
|
|
return fmt.Errorf("source failure changed during replay")
|
|
}
|
|
}
|
|
}
|
|
if cursor == 0 {
|
|
break
|
|
}
|
|
}
|
|
remaining, err := r.redis.Do(ctx, "HLEN", failedKey)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if remaining.Int64() == 0 {
|
|
break
|
|
}
|
|
}
|
|
// Persist producerDone only after every source record was confirmed.
|
|
if err := store.MarkProducerDone(ctx, targetID); err != nil {
|
|
return err
|
|
}
|
|
if _, err := r.redis.Do(ctx, "HSET", sourceKey, "replayState", "completed", "replayError", ""); err != nil {
|
|
return err
|
|
}
|
|
// The source summary has already been archived. Keep the small source
|
|
// metadata for lookup but restore a bounded retention after handoff.
|
|
if _, err := r.redis.Do(ctx, "EXPIRE", sourceKey, 30*24*3600); err != nil {
|
|
return err
|
|
}
|
|
_, err = r.redis.Do(ctx, "SREM", store.Prefix()+"replay:pending", sourceID)
|
|
return err
|
|
}
|
|
|
|
const replayAttention = `
|
|
if redis.call('GET',KEYS[2])~=ARGV[1] then return 0 end
|
|
redis.call('HSET',KEYS[1],'replayState','needs_attention','replayError','Replay handoff stopped; check Worker logs')
|
|
return 1
|
|
`
|
|
|
|
const reserveReplay = `
|
|
local id=redis.call('HGET',KEYS[1],'replayTargetId')
|
|
if id then return id end
|
|
local status=redis.call('HGET',KEYS[1],'status')
|
|
if not status then return 'missing' end
|
|
if status~='completed' and status~='completed_with_failures' then return 'running' end
|
|
if redis.call('HLEN',KEYS[2])==0 then return 'empty' end
|
|
redis.call('HSET',KEYS[1],'replayTargetId',ARGV[1],'replayState','accepted','replayRequestedBy',ARGV[2],'replayEnqueued',0,'replayMaxRound',ARGV[3])
|
|
redis.call('PERSIST',KEYS[1]); redis.call('PERSIST',KEYS[2])
|
|
return ARGV[1]
|
|
`
|
|
const cleanupReplayFailure = `
|
|
if redis.call('HGET',KEYS[1],'replayTargetId')~=ARGV[3] or redis.call('GET',KEYS[3])~=ARGV[4] then return -1 end
|
|
local value=redis.call('HGET',KEYS[2],ARGV[1])
|
|
if not value then return 0 end
|
|
if value~=ARGV[2] then return -1 end
|
|
redis.call('HDEL',KEYS[2],ARGV[1]); redis.call('HINCRBY',KEYS[1],'replayEnqueued',1)
|
|
redis.call('HSET',KEYS[1],'pendingFailureCount',redis.call('HLEN',KEYS[2]))
|
|
return 1
|
|
`
|