chore: 初始开源发布
iotlcm-worker-svr：异步执行中枢（RabbitMQ 消费者 + 全量同步编排）。 - Go module：iotlcm-worker-svr → git.5iot.com/iotlcm-public/iotlcm-worker-svr（214 文件） - 凭据与内部标识脱敏：MySQL/Redis 凭据、私网与生产 IP、内网域名、 库名 xiujie_admin → iotlcm、哨兵名占位 - 159 处真实手机号 → 合法 11 位编造号（保留号段与后 4 位，维持互异， 不破坏「校验 MSISDN 格式」等测试断言） - 移除内部资产：CI 触发器、Jenkins 触发测试文档、预演门禁与镜像 digest 记录 - Dockerfile 改用公共基础镜像（golang:1.25-alpine → alpine:3.23）， 移除写死的 git 凭据 - README 重写（架构/全量同步模型/配置/已知失败），补 Apache-2.0 LICENSE，LF 行尾 + .gitattributes 已知：3 个测试包在源仓同一提交上同样失败（client/upstream、 service/allocate、service/inbound），属既有缺陷，本仓未引入亦未修复。
This commit is contained in:
MiFaZhan
@@ -0,0 +1,493 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
||||
)
|
||||
|
||||
var errProducerDone = gerror.New("run production already closed")
|
||||
|
||||
func producerStateError(state zset.ProducerState) error {
|
||||
if state.Status == "completed" || state.Status == "completed_with_failures" {
|
||||
return ErrRunFinished
|
||||
}
|
||||
if state.ProducerDone {
|
||||
return errProducerDone
|
||||
}
|
||||
if state.Status != "producing" && state.Status != "running" {
|
||||
return gerror.Newf("unexpected producer run status %q", state.Status)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// produceChannels owns all per-attempt coordination. There is one worker per
|
||||
// target, NOT a pool of four long-lived channel workers. SQL tokens and the
|
||||
// cancellable admission gate never span sleeps or goroutine joins.
|
||||
func (p *Producer) produceChannels(ctx context.Context, message model.TrafficUsageMonthlyStartMessage, scope *model.SyncScope, scanUpperID int64, startedAt time.Time, total *productionStats) error {
|
||||
scanner := p.scanner.(channelScanner)
|
||||
metadata, err := scanner.ListChannels(ctx)
|
||||
if err != nil {
|
||||
return gerror.Wrap(err, "initialize channel scan")
|
||||
}
|
||||
state, err := p.store.ReadProducerState(ctx, message.RunID, "")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err = producerStateError(state); err != nil {
|
||||
return err
|
||||
}
|
||||
targets := make(map[string]channelMetadata, len(metadata))
|
||||
known := make(map[string]bool, len(metadata))
|
||||
for _, channel := range metadata {
|
||||
if channel.ChanKey == "" || !scope.Allows(channel.ChanKey) {
|
||||
continue
|
||||
}
|
||||
if known[channel.ChanKey] {
|
||||
return gerror.Newf("duplicate channel metadata %q", channel.ChanKey)
|
||||
}
|
||||
known[channel.ChanKey] = true
|
||||
reason := ""
|
||||
if channel.SyncEnabled {
|
||||
reason = "sync_disabled"
|
||||
} else if channel.TempKey == "" {
|
||||
reason = "missing_channel_template"
|
||||
}
|
||||
if reason != "" {
|
||||
total.skippedChannels++
|
||||
g.Log().Warningf(ctx, "event=monthly_usage_channel status=skipped runId=%s chanKey=%s reason=%s skipUnit=channel", message.RunID, channel.ChanKey, reason)
|
||||
continue
|
||||
}
|
||||
targets[channel.ChanKey] = channel
|
||||
}
|
||||
if scope != nil {
|
||||
for _, chanKey := range scope.Channels {
|
||||
if !known[chanKey] {
|
||||
total.skippedChannels++
|
||||
g.Log().Warningf(ctx, "event=monthly_usage_channel status=skipped runId=%s chanKey=%s reason=unknown_channel skipUnit=channel", message.RunID, chanKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
g.Log().Infof(ctx, "event=monthly_usage_channel_scan status=started runId=%s targetChannels=%d scanUpperID=%d scanConcurrency=%d pageSize=%d skipUnit=channel", message.RunID, len(targets), scanUpperID, p.config.ScanConcurrency, p.config.PageSize)
|
||||
keys := make([]string, 0, len(targets))
|
||||
for key := range targets {
|
||||
keys = append(keys, key)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
finishObservation := p.startObservation(ctx, message.RunID, keys, startedAt)
|
||||
observationSuccess := false
|
||||
defer func() { finishObservation(observationSuccess) }()
|
||||
var (
|
||||
workers sync.WaitGroup
|
||||
slots = make(chan struct{}, p.config.ScanConcurrency)
|
||||
gate = make(chan struct{}, 1)
|
||||
metrics = newScanActivity()
|
||||
results = make([]*channelScanStats, len(keys))
|
||||
)
|
||||
pool, hasPool := scanner.(interface {
|
||||
poolWait(context.Context) (int64, int64)
|
||||
})
|
||||
var poolCount, poolNanos int64
|
||||
if hasPool {
|
||||
poolCount, poolNanos = pool.poolWait(ctx)
|
||||
}
|
||||
workCtx, cancel := context.WithCancelCause(ctx)
|
||||
defer cancel(nil)
|
||||
for i, key := range keys {
|
||||
stats := &channelScanStats{chanKey: key, lastSummary: time.Now()}
|
||||
results[i] = stats
|
||||
workers.Add(1)
|
||||
go func(channel channelMetadata, stats *channelScanStats) {
|
||||
defer workers.Done()
|
||||
scanErr := p.scanOneChannel(workCtx, scanner, message, channel, scanUpperID, startedAt, slots, gate, metrics, stats)
|
||||
if scanErr != nil {
|
||||
cancel(scanErr)
|
||||
}
|
||||
reason := stats.stopReason
|
||||
if reason == "" {
|
||||
reason = "scan_complete"
|
||||
}
|
||||
if scanErr != nil {
|
||||
reason = "cancelled"
|
||||
}
|
||||
p.observe(channel.ChanKey, reason, 0, false, true)
|
||||
stats.log(workCtx, message.RunID, "ended", startedAt, scanErr)
|
||||
}(targets[key], stats)
|
||||
}
|
||||
workers.Wait()
|
||||
var summary channelScanStats
|
||||
for _, stats := range results {
|
||||
total.pages += stats.scanPages
|
||||
total.scanned += stats.eligible
|
||||
total.eligible += stats.eligible
|
||||
total.enqueued += stats.added
|
||||
total.stoppedTasks += stats.rejected
|
||||
if stats.eligible > 0 {
|
||||
total.channels[stats.chanKey] = struct{}{}
|
||||
}
|
||||
summary.merge(stats)
|
||||
}
|
||||
metrics.log(ctx, message.RunID, &summary)
|
||||
if hasPool {
|
||||
count, nanos := pool.poolWait(ctx)
|
||||
g.Log().Infof(ctx, "event=monthly_usage_scan_pool runId=%s sharedPoolWaitCountDelta=%d sharedPoolWaitMsDelta=%d attribution=shared_pool", message.RunID, count-poolCount, (nanos-poolNanos)/int64(time.Millisecond))
|
||||
}
|
||||
observationSuccess = context.Cause(workCtx) == nil
|
||||
return context.Cause(workCtx)
|
||||
}
|
||||
|
||||
func acquireContext(ctx context.Context, token chan struct{}) error {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case token <- struct{}{}:
|
||||
if err := ctx.Err(); err != nil {
|
||||
<-token
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (p *Producer) scanOneChannel(ctx context.Context, scanner channelScanner, message model.TrafficUsageMonthlyStartMessage, channel channelMetadata, scanUpperID int64, startedAt time.Time, slots, gate chan struct{}, metrics *scanActivity, stats *channelScanStats) error {
|
||||
for stats.cursor < scanUpperID {
|
||||
// Detect a blocked/closed run before querying, including empty channels.
|
||||
p.observe(channel.ChanKey, "redis_read", 0, false, false)
|
||||
state, err := p.store.ReadProducerState(ctx, message.RunID, channel.ChanKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err = producerStateError(state); err != nil {
|
||||
return err
|
||||
}
|
||||
if state.Blocked {
|
||||
// No page was read: the unknown unscanned remainder has no card
|
||||
// count. Never turn it into a fabricated success/failure total.
|
||||
stats.stopReason = "blocked_unscanned_remainder"
|
||||
return nil
|
||||
}
|
||||
p.observe(channel.ChanKey, "sql_slot", 0, false, false)
|
||||
waitStart := time.Now()
|
||||
if err = acquireContext(ctx, slots); err != nil {
|
||||
return err
|
||||
}
|
||||
stats.slotWait += time.Since(waitStart)
|
||||
metrics.begin()
|
||||
p.observe(channel.ChanKey, "sql_query", 0, false, false)
|
||||
queryStart := time.Now()
|
||||
cards, scanErr := scanner.ScanChannel(ctx, channel.ChanKey, stats.cursor, p.config.PageSize)
|
||||
elapsed := time.Since(queryStart)
|
||||
metrics.end()
|
||||
<-slots // release immediately after results have been materialized
|
||||
stats.sqlTime += elapsed
|
||||
stats.sqlLatency.observe(elapsed)
|
||||
stats.scanPages++
|
||||
if scanErr != nil {
|
||||
return gerror.Wrapf(scanErr, "scan channel %s after %d", channel.ChanKey, stats.cursor)
|
||||
}
|
||||
if len(cards) > p.config.PageSize {
|
||||
return gerror.New("scanner exceeded page size")
|
||||
}
|
||||
if len(cards) == 0 {
|
||||
stats.stopReason = "empty_result"
|
||||
return nil
|
||||
}
|
||||
var (
|
||||
pageCursor = stats.cursor
|
||||
overUpper bool
|
||||
tasks = make([]zset.Task, 0, len(cards))
|
||||
)
|
||||
for i, card := range cards {
|
||||
if card.ChanKey != channel.ChanKey || card.ID <= pageCursor {
|
||||
return gerror.New("channel scanner returned unordered or foreign cards")
|
||||
}
|
||||
if card.ID > scanUpperID {
|
||||
stats.overUpper += int64(len(cards) - i)
|
||||
overUpper = true
|
||||
break
|
||||
}
|
||||
pageCursor = card.ID
|
||||
tasks = append(tasks, zset.Task{TaskID: zset.StableTaskID(message.RunID, message.Operation, card.ID), RunID: message.RunID, Operation: message.Operation, QueryMonth: message.QueryMonth, ChanKey: channel.ChanKey, TempKey: channel.TempKey, Operator: channel.Operator, CardID: card.ID, ICCID: card.ICCID, MSISDN: card.MSISDN, IMSI: card.IMSI})
|
||||
}
|
||||
p.observe(channel.ChanKey, "page_ready", len(tasks), false, false)
|
||||
stats.eligible += int64(len(tasks))
|
||||
blocked, err := p.admitChannelPage(ctx, message.RunID, tasks, startedAt, gate, stats)
|
||||
if err != nil {
|
||||
return err // page cursor is not committed, even after a partial write
|
||||
}
|
||||
if blocked {
|
||||
stats.stopReason = "blocked"
|
||||
return nil
|
||||
}
|
||||
stats.cursor = pageCursor
|
||||
stats.progress(ctx, message.RunID, startedAt)
|
||||
if overUpper || stats.cursor >= scanUpperID {
|
||||
stats.stopReason = "upper_bound"
|
||||
return nil
|
||||
}
|
||||
// Do not use a short page as EOF: query-time eligibility may change.
|
||||
}
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
// admitChannelPage admits only new TaskIDs, in capacity-sized slices. Each
|
||||
// worker retains at most this page. No capacity reservation survives the gate.
|
||||
func (p *Producer) admitChannelPage(ctx context.Context, runID string, tasks []zset.Task, startedAt time.Time, gate chan struct{}, stats *channelScanStats) (bool, error) {
|
||||
for len(tasks) > 0 {
|
||||
p.observe(stats.chanKey, "admission_gate", len(tasks), false, false)
|
||||
gateStart := time.Now()
|
||||
if err := acquireContext(ctx, gate); err != nil {
|
||||
return false, err
|
||||
}
|
||||
stats.gateWait += time.Since(gateStart)
|
||||
blocked, waitKind, remaining, err := p.admitChannelSlice(ctx, runID, tasks, startedAt, stats)
|
||||
<-gate
|
||||
if err != nil || blocked {
|
||||
return blocked, err
|
||||
}
|
||||
tasks = remaining
|
||||
p.observe(stats.chanKey, "page_ready", len(tasks), false, false)
|
||||
if waitKind != "" {
|
||||
p.observe(stats.chanKey, waitKind+"_capacity", len(tasks), false, false)
|
||||
waitStart := time.Now()
|
||||
err = sleepContext(ctx, p.config.PollInterval)
|
||||
if waitKind == "global" {
|
||||
stats.globalWait += time.Since(waitStart)
|
||||
} else {
|
||||
stats.channelWait += time.Since(waitStart)
|
||||
}
|
||||
stats.progress(ctx, runID, startedAt)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
}
|
||||
return false, ctx.Err()
|
||||
}
|
||||
|
||||
// Caller holds the short admission gate; there is no sleep in this method.
|
||||
func (p *Producer) admitChannelSlice(ctx context.Context, runID string, tasks []zset.Task, startedAt time.Time, stats *channelScanStats) (blocked bool, waitKind string, remaining []zset.Task, err error) {
|
||||
p.observe(stats.chanKey, "redis_read", len(tasks), false, false)
|
||||
readStart := time.Now()
|
||||
state, err := p.store.ReadProducerState(ctx, runID, stats.chanKey)
|
||||
if err != nil {
|
||||
return false, "", tasks, err
|
||||
}
|
||||
if err = producerStateError(state); err != nil {
|
||||
return false, "", tasks, err
|
||||
}
|
||||
existing, err := p.store.ExistingRunTasks(ctx, runID, tasks)
|
||||
stats.stateTime += time.Since(readStart)
|
||||
if err != nil {
|
||||
return false, "", tasks, err
|
||||
}
|
||||
pending := tasks[:0]
|
||||
for i, exists := range existing {
|
||||
if exists {
|
||||
stats.duplicates++
|
||||
} else {
|
||||
pending = append(pending, tasks[i])
|
||||
}
|
||||
}
|
||||
tasks = pending
|
||||
p.observe(stats.chanKey, "redis_read", len(tasks), false, false)
|
||||
if state.Blocked {
|
||||
stats.rejected += int64(len(tasks))
|
||||
return true, "", nil, nil
|
||||
}
|
||||
if len(tasks) == 0 {
|
||||
return false, "", nil, nil
|
||||
}
|
||||
space := int64(len(tasks))
|
||||
if limit := p.config.GlobalHighWaterMark; limit > 0 {
|
||||
space = min(space, limit-state.ActiveCount) // inflight is already included
|
||||
if space <= 0 {
|
||||
return false, "global", tasks, nil
|
||||
}
|
||||
}
|
||||
if limit := p.config.ChannelHighWaterMark; limit > 0 {
|
||||
key, keyErr := p.store.RoundActiveKey(runID, state.CurrentRound, stats.chanKey)
|
||||
if keyErr != nil {
|
||||
return false, "", tasks, keyErr
|
||||
}
|
||||
readStart = time.Now()
|
||||
size, sizeErr := p.store.SizeKey(ctx, key)
|
||||
stats.stateTime += time.Since(readStart)
|
||||
if sizeErr != nil {
|
||||
return false, "", tasks, sizeErr
|
||||
}
|
||||
space = min(space, limit-size)
|
||||
if space <= 0 {
|
||||
return false, "channel", tasks, nil
|
||||
}
|
||||
}
|
||||
batch := tasks[:int(space)]
|
||||
p.observe(stats.chanKey, "redis_write", len(tasks), false, false)
|
||||
enqueueStart := time.Now()
|
||||
added, err := p.store.EnqueueRunBatch(ctx, runID, stats.chanKey, batch, time.Now())
|
||||
stats.enqueueTime += time.Since(enqueueStart)
|
||||
stats.enqueueBatches++
|
||||
if err != nil {
|
||||
return false, "", tasks, err // uncertain responses require idempotent rescan
|
||||
}
|
||||
if added < 0 || added > len(batch) {
|
||||
return false, "", tasks, gerror.New("invalid enqueue count")
|
||||
}
|
||||
stats.added += int64(added)
|
||||
p.observe(stats.chanKey, "redis_write", len(tasks)-len(batch), added > 0, false)
|
||||
if added > 0 && !stats.firstEnqueueRecorded {
|
||||
stats.firstEnqueueRecorded = true
|
||||
stats.firstEnqueue = time.Since(startedAt)
|
||||
stats.log(ctx, runID, "first_enqueue", startedAt, nil)
|
||||
}
|
||||
if added < len(batch) {
|
||||
readStart = time.Now()
|
||||
confirmed, checkErr := p.store.ExistingRunTasks(ctx, runID, batch)
|
||||
if checkErr != nil {
|
||||
return false, "", tasks, checkErr
|
||||
}
|
||||
state, checkErr = p.store.ReadProducerState(ctx, runID, stats.chanKey)
|
||||
stats.stateTime += time.Since(readStart)
|
||||
if checkErr != nil {
|
||||
return false, "", tasks, checkErr
|
||||
}
|
||||
if checkErr = producerStateError(state); checkErr != nil {
|
||||
return false, "", tasks, checkErr
|
||||
}
|
||||
present := 0
|
||||
for _, exists := range confirmed {
|
||||
if exists {
|
||||
present++
|
||||
}
|
||||
}
|
||||
if present < added {
|
||||
return false, "", tasks, gerror.New("enqueue count exceeds confirmed task states")
|
||||
}
|
||||
stats.duplicates += int64(present - added)
|
||||
if state.Blocked {
|
||||
stats.rejected += int64(len(tasks) - present)
|
||||
return true, "", nil, nil
|
||||
}
|
||||
if present != len(batch) {
|
||||
return false, "", tasks, gerror.Newf("unexplained enqueue rejection: added=%d submitted=%d confirmed=%d", added, len(batch), present)
|
||||
}
|
||||
}
|
||||
return false, "", tasks[len(batch):], nil
|
||||
}
|
||||
|
||||
// Retained only to keep errors.Is semantics visible at the call site.
|
||||
func isProductionClosed(err error) bool { return errors.Is(err, errProducerDone) }
|
||||
|
||||
func durationMS(d time.Duration) float64 { return float64(d) / float64(time.Millisecond) }
|
||||
|
||||
// latencyHistogram is fixed-size. Reported percentiles are bucket upper bounds,
|
||||
// never exact samples or a server-side SLA; the last bucket is unbounded.
|
||||
var latencyBounds = [...]time.Duration{time.Millisecond, 2 * time.Millisecond, 5 * time.Millisecond, 10 * time.Millisecond, 20 * time.Millisecond, 50 * time.Millisecond, 100 * time.Millisecond, 200 * time.Millisecond, 500 * time.Millisecond, time.Second, 2 * time.Second, 5 * time.Second, 10 * time.Second, 30 * time.Second, time.Minute}
|
||||
|
||||
type latencyHistogram struct{ counts [16]int64 }
|
||||
|
||||
func (h *latencyHistogram) observe(d time.Duration) {
|
||||
i := 0
|
||||
for i < len(latencyBounds) && d > latencyBounds[i] {
|
||||
i++
|
||||
}
|
||||
h.counts[i]++
|
||||
}
|
||||
func (h *latencyHistogram) percentile(percent int64) string {
|
||||
var total, cumulative int64
|
||||
for _, count := range h.counts {
|
||||
total += count
|
||||
}
|
||||
if total == 0 {
|
||||
return "n/a"
|
||||
}
|
||||
target := (total*percent + 99) / 100
|
||||
for i, count := range h.counts {
|
||||
cumulative += count
|
||||
if cumulative >= target {
|
||||
if i == len(latencyBounds) {
|
||||
return ">60000"
|
||||
}
|
||||
return fmt.Sprintf("%.0f", durationMS(latencyBounds[i]))
|
||||
}
|
||||
}
|
||||
return "n/a"
|
||||
}
|
||||
|
||||
type channelScanStats struct {
|
||||
firstEnqueueRecorded bool
|
||||
chanKey, stopReason string
|
||||
cursor, scanPages, eligible, overUpper, enqueueBatches, added, duplicates, rejected int64
|
||||
sqlTime, slotWait, gateWait, globalWait, channelWait, stateTime, enqueueTime, firstEnqueue time.Duration
|
||||
sqlLatency latencyHistogram
|
||||
lastSummary time.Time
|
||||
}
|
||||
|
||||
func (s *channelScanStats) log(ctx context.Context, runID, status string, startedAt time.Time, err error) {
|
||||
errorText := ""
|
||||
if err != nil {
|
||||
errorText = err.Error()
|
||||
}
|
||||
g.Log().Infof(ctx, "event=monthly_usage_channel_scan status=%s runId=%s chanKey=%s cursor=%d scanPages=%d eligible=%d overUpper=%d enqueueBatches=%d added=%d duplicates=%d rejected=%d sqlMs=%.3f sqlP50UpperMs=%s sqlP95UpperMs=%s slotWaitMs=%.3f admissionWaitMs=%.3f globalCapacityWaitMs=%.3f channelCapacityWaitMs=%.3f stateReadMs=%.3f enqueueMs=%.3f firstEnqueueMs=%.3f wallMs=%d stopReason=%s error=%q", status, runID, s.chanKey, s.cursor, s.scanPages, s.eligible, s.overUpper, s.enqueueBatches, s.added, s.duplicates, s.rejected, durationMS(s.sqlTime), s.sqlLatency.percentile(50), s.sqlLatency.percentile(95), durationMS(s.slotWait), durationMS(s.gateWait), durationMS(s.globalWait), durationMS(s.channelWait), durationMS(s.stateTime), durationMS(s.enqueueTime), durationMS(s.firstEnqueue), time.Since(startedAt).Milliseconds(), s.stopReason, errorText)
|
||||
}
|
||||
func (s *channelScanStats) progress(ctx context.Context, runID string, startedAt time.Time) {
|
||||
if time.Since(s.lastSummary) >= time.Minute {
|
||||
s.lastSummary = time.Now()
|
||||
s.log(ctx, runID, "progress", startedAt, nil)
|
||||
}
|
||||
}
|
||||
func (s *channelScanStats) merge(other *channelScanStats) {
|
||||
s.scanPages += other.scanPages
|
||||
s.eligible += other.eligible
|
||||
s.overUpper += other.overUpper
|
||||
s.enqueueBatches += other.enqueueBatches
|
||||
s.added += other.added
|
||||
s.duplicates += other.duplicates
|
||||
s.rejected += other.rejected
|
||||
s.sqlTime += other.sqlTime
|
||||
s.slotWait += other.slotWait
|
||||
s.gateWait += other.gateWait
|
||||
s.globalWait += other.globalWait
|
||||
s.channelWait += other.channelWait
|
||||
s.stateTime += other.stateTime
|
||||
s.enqueueTime += other.enqueueTime
|
||||
for i, count := range other.sqlLatency.counts {
|
||||
s.sqlLatency.counts[i] += count
|
||||
}
|
||||
}
|
||||
|
||||
type scanActivity struct {
|
||||
mu sync.Mutex
|
||||
started, changed time.Time
|
||||
running, peak int
|
||||
area time.Duration
|
||||
}
|
||||
|
||||
func newScanActivity() *scanActivity {
|
||||
now := time.Now()
|
||||
return &scanActivity{started: now, changed: now}
|
||||
}
|
||||
func (s *scanActivity) change(delta int) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
now := time.Now()
|
||||
s.area += time.Duration(s.running) * now.Sub(s.changed)
|
||||
s.changed = now
|
||||
s.running += delta
|
||||
s.peak = max(s.peak, s.running)
|
||||
}
|
||||
func (s *scanActivity) begin() { s.change(1) }
|
||||
func (s *scanActivity) end() { s.change(-1) }
|
||||
func (s *scanActivity) log(ctx context.Context, runID string, stats *channelScanStats) {
|
||||
wall := time.Since(s.started)
|
||||
g.Log().Infof(ctx, "event=monthly_usage_scan_summary runId=%s wallMs=%d sqlInFlightPeak=%d sqlInFlightAvg=%.4f scanPages=%d enqueueBatches=%d eligible=%d overUpper=%d added=%d duplicates=%d rejected=%d sqlCumulativeMs=%.3f sqlP50UpperMs=%s sqlP95UpperMs=%s slotWaitCumulativeMs=%.3f admissionWaitCumulativeMs=%.3f globalCapacityWaitCumulativeMs=%.3f channelCapacityWaitCumulativeMs=%.3f stateReadCumulativeMs=%.3f enqueueCumulativeMs=%.3f", runID, wall.Milliseconds(), s.peak, float64(s.area)/float64(wall), stats.scanPages, stats.enqueueBatches, stats.eligible, stats.overUpper, stats.added, stats.duplicates, stats.rejected, durationMS(stats.sqlTime), stats.sqlLatency.percentile(50), stats.sqlLatency.percentile(95), durationMS(stats.slotWait), durationMS(stats.gateWait), durationMS(stats.globalWait), durationMS(stats.channelWait), durationMS(stats.stateTime), durationMS(stats.enqueueTime))
|
||||
}
|
||||
@@ -0,0 +1,522 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/testutil"
|
||||
"github.com/gogf/gf/v2/container/gvar"
|
||||
)
|
||||
|
||||
// datasetScanner applies the SQL eligibility predicate before LIMIT; metadata
|
||||
// eligibility is applied by the producer. Optional hooks simulate SQL latency,
|
||||
// live additions and failures without accessing an application database.
|
||||
type datasetScanner struct {
|
||||
cards []Card
|
||||
channels []channelMetadata
|
||||
upper int64
|
||||
hook func(context.Context, string, int64) error
|
||||
queries atomic.Int32
|
||||
lists atomic.Int32
|
||||
inflight atomic.Int32
|
||||
peak atomic.Int32
|
||||
}
|
||||
|
||||
func (s *datasetScanner) SnapshotMaxID(context.Context) (int64, error) { return s.upper, nil }
|
||||
func (s *datasetScanner) ListChannels(context.Context) ([]channelMetadata, error) {
|
||||
s.lists.Add(1)
|
||||
return s.channels, nil
|
||||
}
|
||||
func (s *datasetScanner) Scan(ctx context.Context, last, upper int64, size int) ([]Card, error) {
|
||||
var rows []Card
|
||||
for _, card := range s.cards {
|
||||
if card.ID > last && card.ID <= upper {
|
||||
for _, channel := range s.channels {
|
||||
if channel.ChanKey == card.ChanKey {
|
||||
card.TempKey = channel.TempKey
|
||||
card.SyncEnabled = channel.SyncEnabled
|
||||
card.Operator = channel.Operator
|
||||
}
|
||||
}
|
||||
rows = append(rows, card)
|
||||
if len(rows) == size {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
func (s *datasetScanner) ScanChannel(ctx context.Context, key string, last int64, size int) ([]Card, error) {
|
||||
s.queries.Add(1)
|
||||
running := s.inflight.Add(1)
|
||||
defer s.inflight.Add(-1)
|
||||
for {
|
||||
old := s.peak.Load()
|
||||
if running <= old || s.peak.CompareAndSwap(old, running) {
|
||||
break
|
||||
}
|
||||
}
|
||||
if s.hook != nil {
|
||||
if err := s.hook(ctx, key, last); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
var rows []Card
|
||||
for _, card := range s.cards {
|
||||
if card.ChanKey == key && card.ID > last {
|
||||
rows = append(rows, card)
|
||||
if len(rows) == size {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
return rows, nil
|
||||
}
|
||||
func scannerFor(cards []Card, upper int64) *datasetScanner {
|
||||
keys := map[string]bool{}
|
||||
s := &datasetScanner{cards: cards, upper: upper}
|
||||
for _, card := range cards {
|
||||
if card.ChanKey != "" && !keys[card.ChanKey] {
|
||||
keys[card.ChanKey] = true
|
||||
s.channels = append(s.channels, channelMetadata{ChanKey: card.ChanKey, TempKey: "sim_yd_v5", Operator: 1})
|
||||
}
|
||||
}
|
||||
return s
|
||||
}
|
||||
func testStore(t *testing.T) *zset.Store {
|
||||
t.Helper()
|
||||
store, err := zset.New(testutil.Redis(t), "flow:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return store
|
||||
}
|
||||
func testProducer(t *testing.T, scanner CardScanner, store *zset.Store, config Config) *Producer {
|
||||
t.Helper()
|
||||
config.RoundRetryEnabled = true
|
||||
p, err := New(scanner, store, config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
func contextForTest(t *testing.T) context.Context {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
return ctx
|
||||
}
|
||||
func taskIDs(t *testing.T, store *zset.Store, run string) []string {
|
||||
t.Helper()
|
||||
key, _ := store.TaskStateKey(run)
|
||||
result, err := store.RedisClient().Do(context.Background(), "HKEYS", key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ids := result.Strings()
|
||||
sort.Strings(ids)
|
||||
return ids
|
||||
}
|
||||
func requireMeta(t *testing.T, store *zset.Store, run string, done bool, active int64) {
|
||||
t.Helper()
|
||||
meta, err := store.GetRunMeta(context.Background(), run)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if meta.ProducerDone != done || meta.ActiveCount != active {
|
||||
t.Fatalf("meta=%+v, want done=%t active=%d", meta, done, active)
|
||||
}
|
||||
}
|
||||
func waitUntil(t *testing.T, ctx context.Context, condition func() bool) {
|
||||
t.Helper()
|
||||
for !condition() {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
t.Fatal(ctx.Err())
|
||||
case <-time.After(time.Millisecond):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelModeSettings(t *testing.T) {
|
||||
store, _ := zset.New(&producerRedis{}, "flow:")
|
||||
p, err := New(&fakeScanner{}, store, Config{})
|
||||
if err != nil || p.config.ScanConcurrency != 4 {
|
||||
t.Fatalf("defaults: p=%+v err=%v", p, err)
|
||||
}
|
||||
for _, config := range []Config{{ScanConcurrency: -1}} {
|
||||
if _, err = New(&fakeScanner{}, store, config); err == nil {
|
||||
t.Fatalf("accepted invalid config/capability %+v", config)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelScanFiniteMemoryBoundary(t *testing.T) {
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
upper int64
|
||||
cards []Card
|
||||
ids []int64
|
||||
queries int32
|
||||
}{
|
||||
{"zero upper", 0, []Card{{ID: 1, ChanKey: "a"}}, nil, 0},
|
||||
{"exact page", 4, []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "a"}}, []int64{1, 2, 3, 4}, 2},
|
||||
{"tail then empty", 9, []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}}, []int64{1, 2, 3}, 3},
|
||||
{"first outside", 3, []Card{{ID: 4, ChanKey: "a"}, {ID: 5, ChanKey: "a"}}, nil, 1},
|
||||
{"mixed final page", 3, []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "a"}, {ID: 5, ChanKey: "a"}}, []int64{1, 2, 3}, 2},
|
||||
{"empty channel", 9, nil, nil, 1},
|
||||
} {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
store := testStore(t)
|
||||
scanner := scannerFor(test.cards, test.upper)
|
||||
if len(scanner.channels) == 0 {
|
||||
scanner.channels = []channelMetadata{{ChanKey: "a", TempKey: "sim_yd_v5"}}
|
||||
}
|
||||
p := testProducer(t, scanner, store, Config{PageSize: 2})
|
||||
if err := p.Produce(contextForTest(t), scopedMessage("boundary")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var want []string
|
||||
for _, id := range test.ids {
|
||||
want = append(want, zset.StableTaskID("boundary", operation, id))
|
||||
}
|
||||
sort.Strings(want)
|
||||
got := taskIDs(t, store, "boundary")
|
||||
if len(got) != len(want) || (len(want) > 0 && !reflect.DeepEqual(got, want)) {
|
||||
t.Fatalf("got=%v want=%v", got, want)
|
||||
}
|
||||
if scanner.queries.Load() != test.queries {
|
||||
t.Fatalf("queries=%d want=%d", scanner.queries.Load(), test.queries)
|
||||
}
|
||||
requireMeta(t, store, "boundary", true, int64(len(want)))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelScanLiveHigherIDsDoNotExtendRun(t *testing.T) {
|
||||
store := testStore(t)
|
||||
scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}}, 3)
|
||||
scanner.hook = func(_ context.Context, _ string, last int64) error {
|
||||
if last > 0 {
|
||||
scanner.cards = append(scanner.cards, Card{ID: 100 + last, ChanKey: "a"})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
p := testProducer(t, scanner, store, Config{PageSize: 2})
|
||||
if err := p.Produce(contextForTest(t), scopedMessage("live")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(taskIDs(t, store, "live")) != 3 || scanner.queries.Load() != 2 {
|
||||
t.Fatal("higher IDs extended finite range")
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelScanSerialTaskIDSetParity(t *testing.T) {
|
||||
for _, scope := range [][]string{nil, {"a", "b"}, {"disabled", "unknown"}} {
|
||||
t.Run(fmt.Sprint(scope), func(t *testing.T) {
|
||||
store := testStore(t)
|
||||
cards := []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "b"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "disabled"}, {ID: 5, ChanKey: "no-template"}, {ID: 6, ChanKey: "unknown"}, {ID: 7, ChanKey: "b"}, {ID: 8, ChanKey: "a"}}
|
||||
var sets [][]string
|
||||
for _, mode := range []string{"serial", "channel"} {
|
||||
scanner := scannerFor(cards, 7)
|
||||
scanner.channels = []channelMetadata{{ChanKey: "a", TempKey: "sim_yd_v5", Operator: 1}, {ChanKey: "b", TempKey: "sim_dx_v1", Operator: 2}, {ChanKey: "disabled", TempKey: "sim_yd_v5", SyncEnabled: true}, {ChanKey: "no-template"}}
|
||||
p := testProducer(t, scanner, store, Config{PageSize: 2})
|
||||
run := "parity-" + mode
|
||||
if err := p.Produce(contextForTest(t), scopedMessage(run, scope...)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ids := taskIDs(t, store, run)
|
||||
for i, id := range ids {
|
||||
ids[i] = strings.TrimPrefix(id, run+":")
|
||||
}
|
||||
sets = append(sets, ids)
|
||||
if mode == "channel" && scanner.lists.Load() != 1 {
|
||||
t.Fatal("metadata not loaded once")
|
||||
}
|
||||
_, _ = store.RedisClient().Do(context.Background(), "SREM", store.ActiveRunsKey(), run)
|
||||
}
|
||||
if !reflect.DeepEqual(sets[0], sets[1]) {
|
||||
t.Fatalf("serial=%v channel=%v", sets[0], sets[1])
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelSQLSlotsReleasedWhileChannelsWait(t *testing.T) {
|
||||
store := testStore(t)
|
||||
ctx := contextForTest(t)
|
||||
var cards []Card
|
||||
for i := 0; i < 9; i++ {
|
||||
cards = append(cards, Card{ID: int64(i + 1), ChanKey: fmt.Sprint(i)}, Card{ID: int64(i + 10), ChanKey: fmt.Sprint(i)})
|
||||
}
|
||||
sort.Slice(cards, func(i, j int) bool { return cards[i].ID < cards[j].ID })
|
||||
scanner := scannerFor(cards, 100)
|
||||
scanner.hook = func(ctx context.Context, _ string, _ int64) error { return sleepContext(ctx, 5*time.Millisecond) }
|
||||
p := testProducer(t, scanner, store, Config{PageSize: 2, ScanConcurrency: 4, ChannelHighWaterMark: 1, GlobalHighWaterMark: 100, PollInterval: time.Millisecond})
|
||||
result := make(chan error, 1)
|
||||
go func() { result <- p.Produce(ctx, scopedMessage("isolation")) }()
|
||||
waitUntil(t, ctx, func() bool { return len(taskIDs(t, store, "isolation")) == 9 })
|
||||
if peak := scanner.peak.Load(); peak > 4 || peak < 2 {
|
||||
t.Fatalf("SQL peak=%d", peak)
|
||||
}
|
||||
// All nine channels got a first task despite the first four waiting at
|
||||
// channel capacity. SQL slots are free while every channel is paused.
|
||||
if scanner.inflight.Load() != 0 {
|
||||
t.Fatal("SQL tokens held during channel backpressure")
|
||||
}
|
||||
for _, channel := range scanner.channels {
|
||||
if _, err := store.BlockRunScoped(ctx, "isolation", channel.ChanKey); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if err := <-result; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
meta, err := store.GetRunMeta(ctx, "isolation")
|
||||
if err != nil || !meta.ProducerDone {
|
||||
t.Fatalf("meta=%+v err=%v", meta, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelRemainingCapacityAndDuplicateAdmission(t *testing.T) {
|
||||
store := testStore(t)
|
||||
ctx := contextForTest(t)
|
||||
scanner := scannerFor(nil, 0)
|
||||
p := testProducer(t, scanner, store, Config{GlobalHighWaterMark: 3, ChannelHighWaterMark: 2, PollInterval: time.Millisecond})
|
||||
if err := store.CreateRun(ctx, zset.RunMeta{RunID: "capacity", Status: "producing"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Existing success and waiting states still deduplicate and consume no new
|
||||
// capacity. Active includes leases: inflight must not be added again.
|
||||
stateKey, _ := store.TaskStateKey("capacity")
|
||||
metaKey, _ := store.RunMetaKey("capacity")
|
||||
activeKey, _ := store.RoundActiveKey("capacity", 1, "a")
|
||||
_, _ = store.RedisClient().Do(ctx, "HSET", stateKey, "one", `{"status":"success"}`, "two", `{"status":"waiting"}`)
|
||||
_, _ = store.RedisClient().Do(ctx, "HSET", metaKey, "activeCount", 2, "inflightCount", 2)
|
||||
stats := &channelScanStats{chanKey: "a", lastSummary: time.Now()}
|
||||
tasks := []zset.Task{{TaskID: "one", RunID: "capacity", ChanKey: "a"}, {TaskID: "two", RunID: "capacity", ChanKey: "a"}, {TaskID: "three", RunID: "capacity", ChanKey: "a"}, {TaskID: "four", RunID: "capacity", ChanKey: "a"}}
|
||||
gate := make(chan struct{}, 1)
|
||||
result := make(chan error, 1)
|
||||
go func() { _, err := p.admitChannelPage(ctx, "capacity", tasks, time.Now(), gate, stats); result <- err }()
|
||||
waitUntil(t, ctx, func() bool { meta, e := store.GetRunMeta(ctx, "capacity"); return e == nil && meta.ActiveCount == 3 })
|
||||
if size, err := store.SizeKey(ctx, activeKey); err != nil || size != 1 {
|
||||
t.Fatalf("initial slice size=%d err=%v", size, err)
|
||||
}
|
||||
_, _ = store.RedisClient().Do(ctx, "HSET", metaKey, "activeCount", 0, "inflightCount", 0)
|
||||
if err := <-result; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stats.duplicates != 2 || stats.added != 2 || stats.enqueueBatches != 2 || stats.globalWait == 0 {
|
||||
t.Fatalf("stats=%+v", stats)
|
||||
}
|
||||
}
|
||||
|
||||
// faultRedis forwards to a real isolated Redis, optionally returning a lost
|
||||
// response after Lua succeeded. Never mocks the Lua semantics under test.
|
||||
type faultRedis struct {
|
||||
zset.RedisClient
|
||||
failEnqueue atomic.Int32
|
||||
failRead atomic.Bool
|
||||
failRenew atomic.Bool
|
||||
lostResponse bool
|
||||
reject bool
|
||||
}
|
||||
|
||||
func (r *faultRedis) Do(ctx context.Context, command string, args ...any) (*gvar.Var, error) {
|
||||
if r.failRead.Load() && command == "HMGET" {
|
||||
return nil, errors.New("Redis read failure")
|
||||
}
|
||||
return r.RedisClient.Do(ctx, command, args...)
|
||||
}
|
||||
func (r *faultRedis) Eval(ctx context.Context, script string, n int64, keys []string, args []any) (*gvar.Var, error) {
|
||||
if r.failRenew.Load() && strings.Contains(script, "PEXPIRE") && strings.Contains(script, "owner ~=") {
|
||||
return gvar.New(0), nil
|
||||
}
|
||||
if n == 4 && strings.Contains(script, "HEXISTS") {
|
||||
if r.reject {
|
||||
return gvar.New(0), nil
|
||||
}
|
||||
if r.failEnqueue.CompareAndSwap(1, 0) {
|
||||
if r.lostResponse {
|
||||
if _, err := r.RedisClient.Eval(ctx, script, n, keys, args); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return nil, errors.New("enqueue response unavailable")
|
||||
}
|
||||
}
|
||||
return r.RedisClient.Eval(ctx, script, n, keys, args)
|
||||
}
|
||||
|
||||
func TestChannelRescanPartialAndUncertainWrites(t *testing.T) {
|
||||
for _, uncertain := range []bool{false, true} {
|
||||
t.Run(fmt.Sprint(uncertain), func(t *testing.T) {
|
||||
client := &faultRedis{RedisClient: testutil.Redis(t), lostResponse: uncertain}
|
||||
store, _ := zset.New(client, "flow:")
|
||||
scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "a"}, {ID: 3, ChanKey: "a"}, {ID: 4, ChanKey: "a"}}, 4)
|
||||
// Fail the second page: first page must survive; uncertain success is
|
||||
// confirmed only on a same-Run rescan, with stable IDs and no double count.
|
||||
scanner.hook = func(_ context.Context, _ string, last int64) error {
|
||||
if last == 2 {
|
||||
client.failEnqueue.Store(1)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
p := testProducer(t, scanner, store, Config{PageSize: 2})
|
||||
if err := p.Produce(contextForTest(t), scopedMessage("rescan")); err == nil {
|
||||
t.Fatal("expected enqueue failure")
|
||||
}
|
||||
active := int64(2)
|
||||
if uncertain {
|
||||
active = 4
|
||||
}
|
||||
requireMeta(t, store, "rescan", false, active)
|
||||
scanner.hook = nil
|
||||
if err := p.Produce(contextForTest(t), scopedMessage("rescan")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
requireMeta(t, store, "rescan", true, 4)
|
||||
if len(taskIDs(t, store, "rescan")) != 4 {
|
||||
t.Fatal("wrong stable task set")
|
||||
}
|
||||
before := scanner.queries.Load()
|
||||
if err := p.Produce(contextForTest(t), scopedMessage("rescan")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if before != scanner.queries.Load() {
|
||||
t.Fatal("producerDone redelivery rescanned cards")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelFailureCancelsAllWorkersWithoutProducerDone(t *testing.T) {
|
||||
for _, kind := range []string{"sql", "redis", "unexplained_rejection", "lease"} {
|
||||
t.Run(kind, func(t *testing.T) {
|
||||
client := &faultRedis{RedisClient: testutil.Redis(t)}
|
||||
store, _ := zset.New(client, "flow:")
|
||||
cards := []Card{{ID: 1, ChanKey: "a"}, {ID: 2, ChanKey: "b"}, {ID: 3, ChanKey: "c"}}
|
||||
scanner := scannerFor(cards, 10)
|
||||
allStarted := make(chan struct{})
|
||||
var started atomic.Int32
|
||||
var once sync.Once
|
||||
scanner.hook = func(ctx context.Context, key string, _ int64) error {
|
||||
if started.Add(1) >= 3 {
|
||||
once.Do(func() { close(allStarted) })
|
||||
}
|
||||
select {
|
||||
case <-allStarted:
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
if kind == "lease" {
|
||||
client.failRenew.Store(true)
|
||||
<-ctx.Done()
|
||||
return ctx.Err()
|
||||
}
|
||||
if key == "a" {
|
||||
switch kind {
|
||||
case "sql":
|
||||
return errors.New("SQL failed")
|
||||
case "redis":
|
||||
client.failRead.Store(true)
|
||||
case "unexplained_rejection":
|
||||
client.reject = true
|
||||
}
|
||||
return nil
|
||||
}
|
||||
<-ctx.Done()
|
||||
return ctx.Err()
|
||||
}
|
||||
p := testProducer(t, scanner, store, Config{PageSize: 2, RunLockTTL: 90 * time.Millisecond})
|
||||
if err := p.Produce(contextForTest(t), scopedMessage("failure")); err == nil {
|
||||
t.Fatal("expected fatal failure")
|
||||
}
|
||||
if scanner.inflight.Load() != 0 {
|
||||
t.Fatal("scanner goroutines leaked")
|
||||
}
|
||||
meta, err := store.GetRunMeta(context.Background(), "failure")
|
||||
if err != nil || meta.ProducerDone {
|
||||
t.Fatalf("partial scan marked done: %+v err=%v", meta, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelScanStateStopsAndNoTarget(t *testing.T) {
|
||||
for _, kind := range []string{"terminal", "producerDone", "blocked", "missing", "no-target", "no-cards"} {
|
||||
t.Run(kind, func(t *testing.T) {
|
||||
store := testStore(t)
|
||||
scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}}, 2)
|
||||
if kind == "no-target" {
|
||||
scanner.channels[0].SyncEnabled = true
|
||||
}
|
||||
if kind == "no-cards" {
|
||||
scanner.cards = nil
|
||||
}
|
||||
p := testProducer(t, scanner, store, Config{})
|
||||
if kind != "no-target" && kind != "no-cards" {
|
||||
if err := store.CreateRun(context.Background(), zset.RunMeta{RunID: "stop", Status: "producing"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key, _ := store.RunMetaKey("stop")
|
||||
switch kind {
|
||||
case "terminal":
|
||||
_, _ = store.RedisClient().Do(context.Background(), "HSET", key, "status", "completed")
|
||||
case "producerDone":
|
||||
if err := store.MarkProducerDone(context.Background(), "stop"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
case "blocked":
|
||||
_, _ = store.BlockRunScoped(context.Background(), "stop", "a")
|
||||
case "missing":
|
||||
scanner.hook = func(ctx context.Context, _ string, _ int64) error {
|
||||
_, err := store.RedisClient().Do(ctx, "DEL", key)
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
err := p.Produce(contextForTest(t), scopedMessage("stop"))
|
||||
if kind == "terminal" {
|
||||
if !errors.Is(err, ErrRunFinished) {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
} else if kind == "missing" {
|
||||
if !errors.Is(err, zset.ErrRunNotFound) {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
} else if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(taskIDs(t, store, "stop")) != 0 {
|
||||
t.Fatal("stopped/empty run enqueued tasks")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// Omitted retry/scan-mode settings must still use the channel Run producer.
|
||||
func TestDefaultProducerUsesChannelRun(t *testing.T) {
|
||||
store := testStore(t)
|
||||
scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}}, 1)
|
||||
p, err := New(scanner, store, Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err = p.Produce(contextForTest(t), scopedMessage("default-channel")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if scanner.queries.Load() == 0 || scanner.lists.Load() != 1 {
|
||||
t.Fatal("default did not use channel scan")
|
||||
}
|
||||
requireMeta(t, store, "default-channel", true, 1)
|
||||
}
|
||||
@@ -0,0 +1,78 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/gogf/gf/v2/errors/gerror"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/dao"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/entity"
|
||||
)
|
||||
|
||||
// channelScanner augments, rather than replaces, the legacy scanner contract.
|
||||
// ListChannels is called once before any workers start. ScanChannel must not
|
||||
// mutate the legacy DatabaseScanner.channelMetadata cache.
|
||||
type channelScanner interface {
|
||||
ListChannels(context.Context) ([]channelMetadata, error)
|
||||
ScanChannel(context.Context, string, int64, int) ([]Card, error)
|
||||
}
|
||||
|
||||
// This exact MySQL query shape is a release gate. Do not silently remove hints,
|
||||
// add an SQL upper bound, or move eligibility filtering after LIMIT.
|
||||
const channelPageSQL = `SELECT c.id, c.iccid, c.msisdn, c.imsi, c.chan_key
|
||||
FROM t_sim_card_info AS c FORCE INDEX (idx_card_chan_key)
|
||||
STRAIGHT_JOIN sys_dept AS d
|
||||
ON c.dept_id = d.dept_id
|
||||
AND c.tenant_id = d.tenant_id
|
||||
WHERE c.chan_key = ?
|
||||
AND c.id > ?
|
||||
AND c.iccid <> ''
|
||||
AND c.current_package_code <> ''
|
||||
AND d.deleted_at IS NULL
|
||||
ORDER BY c.id ASC
|
||||
LIMIT ?`
|
||||
|
||||
func (s *DatabaseScanner) ListChannels(ctx context.Context) ([]channelMetadata, error) {
|
||||
var metadata []channelMetadata
|
||||
if err := dao.TSimCardInfo.DB().Model("t_sim_channels").Ctx(ctx).Fields("chan_key,temp_key,sync_enabled,operator").Scan(&metadata); err != nil {
|
||||
return nil, gerror.Wrap(err, "list channel metadata")
|
||||
}
|
||||
return metadata, nil
|
||||
}
|
||||
|
||||
func (s *DatabaseScanner) ScanChannel(ctx context.Context, chanKey string, lastID int64, pageSize int) ([]Card, error) {
|
||||
if chanKey == "" || lastID < 0 || pageSize <= 0 {
|
||||
return nil, gerror.New("invalid channel pagination parameters")
|
||||
}
|
||||
var rows []entity.TSimCardInfo
|
||||
// Use the card DAO's database group, not a separately configured connection.
|
||||
if err := dao.TSimCardInfo.DB().Raw(channelPageSQL, chanKey, lastID, pageSize).Ctx(ctx).Scan(&rows); err != nil {
|
||||
return nil, gerror.Wrap(err, "scan channel cards")
|
||||
}
|
||||
cards := make([]Card, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
cards = append(cards, Card{ID: row.Id, ICCID: row.Iccid, MSISDN: row.Msisdn, IMSI: row.Imsi, ChanKey: row.ChanKey})
|
||||
}
|
||||
return cards, nil
|
||||
}
|
||||
|
||||
// Pool counters are shared with other worker traffic. Their deltas are direct
|
||||
// observations, not an attribution of all connection waits to this producer.
|
||||
func (*DatabaseScanner) poolWait(ctx context.Context) (int64, int64) {
|
||||
var count, nanos int64
|
||||
for _, item := range dao.TSimCardInfo.DB().Stats(ctx) {
|
||||
stats := item.Stats()
|
||||
count += stats.WaitCount
|
||||
nanos += int64(stats.WaitDuration)
|
||||
}
|
||||
return count, nanos
|
||||
}
|
||||
|
||||
// ValidateScanSettings is strict for values read from configuration. The Go
|
||||
// constructor first fills omitted zero values; explicit YAML zero is rejected.
|
||||
func ValidateScanSettings(concurrency int) error {
|
||||
if concurrency <= 0 {
|
||||
return gerror.New("trafficUsageFullSync.producer.scanConcurrency must be positive")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/dao"
|
||||
_ "github.com/gogf/gf/contrib/drivers/sqlite/v2"
|
||||
"github.com/gogf/gf/v2/database/gdb"
|
||||
)
|
||||
|
||||
// SQLite runs the SAME predicate and ordering with only the two MySQL hints
|
||||
// removed. This checks full eligibility/TaskID parity, not MySQL access plans.
|
||||
// Production EXPLAIN must still validate the unmodified channelPageSQL.
|
||||
type sqliteChannelScanner struct{ DatabaseScanner }
|
||||
|
||||
func (s *sqliteChannelScanner) ScanChannel(ctx context.Context, key string, last int64, size int) ([]Card, error) {
|
||||
query := strings.ReplaceAll(channelPageSQL, "FORCE INDEX (idx_card_chan_key)", "")
|
||||
query = strings.ReplaceAll(query, "STRAIGHT_JOIN", "INNER JOIN")
|
||||
var rows []struct {
|
||||
ID int64 `orm:"id"`
|
||||
ICCID string `orm:"iccid"`
|
||||
MSISDN string `orm:"msisdn"`
|
||||
IMSI string `orm:"imsi"`
|
||||
ChanKey string `orm:"chan_key"`
|
||||
}
|
||||
if err := dao.TSimCardInfo.DB().Raw(query, key, last, size).Ctx(ctx).Scan(&rows); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cards := make([]Card, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
cards = append(cards, Card{ID: row.ID, ICCID: row.ICCID, MSISDN: row.MSISDN, IMSI: row.IMSI, ChanKey: row.ChanKey})
|
||||
}
|
||||
return cards, nil
|
||||
}
|
||||
|
||||
func TestDatabaseQualificationAndCompleteTaskIDParity(t *testing.T) {
|
||||
// No application DB/config: isolated SQLite file and loopback Redis only.
|
||||
if err := gdb.SetConfig(gdb.Config{"default": gdb.ConfigGroup{{Type: "sqlite", Link: "sqlite::@file(" + t.TempDir() + "/cards.db)"}}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := contextForTest(t)
|
||||
db := dao.TSimCardInfo.DB()
|
||||
t.Cleanup(func() { _ = db.Close(context.Background()) })
|
||||
for _, sql := range []string{
|
||||
`CREATE TABLE sys_dept (dept_id INTEGER PRIMARY KEY, tenant_id TEXT, deleted_at DATETIME)`,
|
||||
`CREATE TABLE t_sim_card_info (id INTEGER PRIMARY KEY, iccid TEXT, msisdn TEXT, imsi TEXT, chan_key TEXT, current_package_code TEXT, dept_id INTEGER, tenant_id TEXT)`,
|
||||
`CREATE INDEX idx_card_chan_key ON t_sim_card_info(chan_key)`,
|
||||
`CREATE TABLE t_sim_channels (chan_key TEXT, temp_key TEXT, sync_enabled INTEGER, operator INTEGER)`,
|
||||
`INSERT INTO sys_dept VALUES (1,'t1',NULL),(2,'t2',NULL),(3,'t1','2026-01-01')`,
|
||||
`INSERT INTO t_sim_channels VALUES ('a','sim_yd_v5',0,1),('b','sim_dx_v1',0,2),('disabled','sim_yd_v5',1,1),('no-template','',0,1)`,
|
||||
`INSERT INTO t_sim_card_info VALUES
|
||||
(1,'i1','m1','s1','a','pkg',1,'t1'),
|
||||
(2,'i2','m2','s2','b','pkg',2,'t2'),
|
||||
(3,'i3','','','a','pkg',1,'t1'),
|
||||
(4,'','','','a','pkg',1,'t1'),
|
||||
(5,'i5','','','a','',1,'t1'),
|
||||
(6,'i6','','','a','pkg',1,'wrong-tenant'),
|
||||
(7,'i7','','','a','pkg',3,'t1'),
|
||||
(8,'i8','','','b','pkg',99,'t2'),
|
||||
(9,'i9','','','disabled','pkg',1,'t1'),
|
||||
(10,'i10','','','no-template','pkg',1,'t1'),
|
||||
(11,'i11','','','unknown','pkg',1,'t1'),
|
||||
(12,'i12','','','','pkg',1,'t1'),
|
||||
(13,'i13','','','b','pkg',2,'t2')`,
|
||||
} {
|
||||
if _, err := db.Exec(ctx, sql); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
store := testStore(t)
|
||||
for _, scope := range [][]string{nil, {"a"}, {"b", "disabled", "no-template", "unknown"}, {"unknown"}} {
|
||||
var sets [][]string
|
||||
for _, mode := range []string{"serial", "channel"} {
|
||||
scanner := &sqliteChannelScanner{}
|
||||
p := testProducer(t, scanner, store, Config{PageSize: 2})
|
||||
run := "db-" + mode
|
||||
if err := p.Produce(ctx, scopedMessage(run, scope...)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ids := taskIDs(t, store, run)
|
||||
for i, id := range ids {
|
||||
ids[i] = strings.TrimPrefix(id, run+":")
|
||||
}
|
||||
sort.Strings(ids)
|
||||
sets = append(sets, ids)
|
||||
stateKey, _ := store.TaskStateKey(run)
|
||||
metaKey, _ := store.RunMetaKey(run)
|
||||
chanKey, _ := store.RunChannelsKey(run)
|
||||
_, _ = store.RedisClient().Do(ctx, "DEL", stateKey, metaKey, chanKey)
|
||||
for _, key := range []string{"a", "b"} {
|
||||
activeKey, _ := store.RoundActiveKey(run, 1, key)
|
||||
_, _ = store.RedisClient().Do(ctx, "DEL", activeKey)
|
||||
}
|
||||
_, _ = store.RedisClient().Do(ctx, "SREM", store.ActiveRunsKey(), run)
|
||||
}
|
||||
if !reflect.DeepEqual(sets[0], sets[1]) {
|
||||
t.Fatalf("scope=%v serial=%v channel=%v", scope, sets[0], sets[1])
|
||||
}
|
||||
if scope == nil {
|
||||
want := []string{operation + ":1", operation + ":13", operation + ":2", operation + ":3"}
|
||||
sort.Strings(want)
|
||||
if !reflect.DeepEqual(sets[0], want) {
|
||||
t.Fatalf("qualification set=%v want=%v", sets[0], want)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestChannelSQLShapeAndParameters(t *testing.T) {
|
||||
for _, term := range []string{"FORCE INDEX (idx_card_chan_key)", "STRAIGHT_JOIN sys_dept", "c.chan_key = ?", "c.id > ?", "c.iccid <> ''", "c.current_package_code <> ''", "d.deleted_at IS NULL", "c.tenant_id = d.tenant_id", "ORDER BY c.id ASC", "LIMIT ?"} {
|
||||
if !strings.Contains(channelPageSQL, term) {
|
||||
t.Fatalf("missing %q", term)
|
||||
}
|
||||
}
|
||||
if strings.Contains(channelPageSQL, "<=") || strings.Count(channelPageSQL, "?") != 3 {
|
||||
t.Fatal("SQL contains upper bound or changed bind parameter count")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,485 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/gogf/gf/v2/database/gdb"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/dao"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/supply"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/entity"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
||||
)
|
||||
|
||||
const operation = "current_month_usage_full_sync"
|
||||
|
||||
// scopeSampleLimit bounds how many channel names are copied into run metadata.
|
||||
// The run hash and its archive JSON must stay bounded regardless of how many
|
||||
// channels a scope selects.
|
||||
const scopeSampleLimit = 10
|
||||
|
||||
// scopeDigest is a stable short fingerprint of the selected channels. It lets
|
||||
// operators confirm two runs covered the same scope without storing the list.
|
||||
func scopeDigest(scope *model.SyncScope) string {
|
||||
if scope == nil || len(scope.Channels) == 0 {
|
||||
return ""
|
||||
}
|
||||
digest := sha256.Sum256([]byte(strings.Join(scope.Channels, "\n")))
|
||||
return hex.EncodeToString(digest[:8])
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
SupplyObservation *supply.Tracker
|
||||
// Empty/zero mean omitted for Go callers; startup config validates explicit values.
|
||||
ScanConcurrency int
|
||||
PageSize int
|
||||
GlobalHighWaterMark int64
|
||||
ChannelHighWaterMark int64
|
||||
PollInterval time.Duration
|
||||
RunLockKey string
|
||||
RunLockTTL time.Duration
|
||||
RoundRetryEnabled bool
|
||||
MaxRounds int
|
||||
FailureRetention time.Duration
|
||||
}
|
||||
|
||||
func (c Config) withDefaults() Config {
|
||||
c.RoundRetryEnabled = true
|
||||
if c.ScanConcurrency == 0 {
|
||||
c.ScanConcurrency = 4
|
||||
}
|
||||
if c.PageSize <= 0 {
|
||||
c.PageSize = 100
|
||||
}
|
||||
if c.PollInterval <= 0 {
|
||||
c.PollInterval = 2 * time.Second
|
||||
}
|
||||
if c.RunLockKey == "" {
|
||||
c.RunLockKey = "flow:producer:active"
|
||||
}
|
||||
if c.RunLockTTL <= 0 {
|
||||
c.RunLockTTL = 5 * time.Minute
|
||||
}
|
||||
if c.MaxRounds <= 0 {
|
||||
c.MaxRounds = 5
|
||||
}
|
||||
if c.FailureRetention <= 0 {
|
||||
c.FailureRetention = 30 * 24 * time.Hour
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
type CardScanner interface {
|
||||
SnapshotMaxID(context.Context) (int64, error)
|
||||
}
|
||||
|
||||
type Card struct {
|
||||
ID int64
|
||||
ICCID string
|
||||
MSISDN string
|
||||
IMSI string
|
||||
ChanKey string
|
||||
TempKey string
|
||||
SyncEnabled bool
|
||||
Operator int
|
||||
}
|
||||
|
||||
type channelMetadata struct {
|
||||
ChanKey string `orm:"chan_key"`
|
||||
TempKey string `orm:"temp_key"`
|
||||
SyncEnabled bool `orm:"sync_enabled"`
|
||||
Operator int `orm:"operator"`
|
||||
}
|
||||
|
||||
type DatabaseScanner struct{}
|
||||
|
||||
func (s *DatabaseScanner) SnapshotMaxID(ctx context.Context) (int64, error) {
|
||||
var card entity.TSimCardInfo
|
||||
cardTable := dao.TSimCardInfo.Table()
|
||||
columns := dao.TSimCardInfo.Columns()
|
||||
if err := monthlyUsageEligibleCardModel(ctx).
|
||||
Fields(cardTable + "." + columns.Id + " AS " + columns.Id).
|
||||
OrderDesc(cardTable + "." + columns.Id).
|
||||
Limit(1).
|
||||
Scan(&card); err != nil {
|
||||
return 0, fmt.Errorf("read card snapshot max id: %w", err)
|
||||
}
|
||||
return card.Id, nil
|
||||
}
|
||||
|
||||
func monthlyUsageEligibleCardModel(ctx context.Context) *gdb.Model {
|
||||
cardTable := dao.TSimCardInfo.Table()
|
||||
deptTable := dao.SysDept.Table()
|
||||
cardColumns := dao.TSimCardInfo.Columns()
|
||||
deptColumns := dao.SysDept.Columns()
|
||||
|
||||
return dao.TSimCardInfo.Ctx(ctx).
|
||||
InnerJoin(deptTable, cardTable+"."+cardColumns.DeptId+" = "+deptTable+"."+deptColumns.DeptId+" AND "+cardTable+"."+cardColumns.TenantId+" = "+deptTable+"."+deptColumns.TenantId).
|
||||
Where(cardTable+"."+cardColumns.Iccid+" <> ?", "").
|
||||
Where(cardTable+"."+cardColumns.ChanKey+" <> ?", "").
|
||||
Where(cardTable+"."+cardColumns.CurrentPackageCode+" <> ?", "").
|
||||
WhereNull(deptTable + "." + deptColumns.DeletedAt)
|
||||
}
|
||||
|
||||
type Producer struct {
|
||||
scanner CardScanner
|
||||
store *zset.Store
|
||||
config Config
|
||||
running atomic.Int32
|
||||
owner string
|
||||
observation *supply.Observer
|
||||
}
|
||||
|
||||
type productionStats struct {
|
||||
pages int64
|
||||
scanned int64
|
||||
eligible int64
|
||||
enqueued int64
|
||||
skippedEmptyChan int64
|
||||
skippedNoTemplate int64
|
||||
skippedSyncDisabled int64
|
||||
skippedOutOfScope int64
|
||||
skippedChannels int64
|
||||
stoppedTasks int64
|
||||
channels map[string]struct{}
|
||||
scopeChannels int
|
||||
}
|
||||
|
||||
func (s *productionStats) channelCount() int {
|
||||
return len(s.channels)
|
||||
}
|
||||
|
||||
func (s *productionStats) log(ctx context.Context, status, runID, queryMonth, stage, reason string, startedAt time.Time, maxID int64, err error) {
|
||||
message := fmt.Sprintf("event=monthly_usage_producer status=%s runId=%s queryMonth=%s scanUpperID=%d pages=%d scanned=%d eligible=%d enqueued=%d skippedEmptyChan=%d skippedNoTemplate=%d skippedSyncDisabled=%d skippedOutOfScope=%d stoppedTasks=%d channelCount=%d scopeChannels=%d durationMs=%d", status, runID, queryMonth, maxID, s.pages, s.scanned, s.eligible, s.enqueued, s.skippedEmptyChan, s.skippedNoTemplate, s.skippedSyncDisabled, s.skippedOutOfScope, s.stoppedTasks, s.channelCount(), s.scopeChannels, time.Since(startedAt).Milliseconds())
|
||||
message += fmt.Sprintf(" skippedChannels=%d", s.skippedChannels)
|
||||
if s.scanned > 0 {
|
||||
message += fmt.Sprintf(" cardsPerSec=%.2f", float64(s.scanned)/time.Since(startedAt).Seconds())
|
||||
}
|
||||
if stage != "" {
|
||||
message += " stage=" + stage
|
||||
}
|
||||
if reason != "" {
|
||||
message += " reason=" + reason
|
||||
}
|
||||
if err != nil {
|
||||
message += fmt.Sprintf(" error=%q", err.Error())
|
||||
}
|
||||
if status == "failed" {
|
||||
g.Log().Error(ctx, message)
|
||||
return
|
||||
}
|
||||
if status == "skipped" {
|
||||
g.Log().Warning(ctx, message)
|
||||
return
|
||||
}
|
||||
g.Log().Info(ctx, message)
|
||||
}
|
||||
|
||||
func New(scanner CardScanner, store *zset.Store, config Config) (*Producer, error) {
|
||||
if scanner == nil || store == nil {
|
||||
return nil, fmt.Errorf("monthly usage producer dependencies are incomplete")
|
||||
}
|
||||
config = config.withDefaults()
|
||||
if err := ValidateScanSettings(config.ScanConcurrency); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if config.RoundRetryEnabled {
|
||||
if _, ok := scanner.(channelScanner); !ok {
|
||||
return nil, fmt.Errorf("channel scan requires ListChannels and ScanChannel capabilities")
|
||||
}
|
||||
}
|
||||
return &Producer{scanner: scanner, store: store, config: config, owner: newOwner()}, nil
|
||||
}
|
||||
|
||||
var ErrAlreadyRunning = fmt.Errorf("monthly usage producer already running")
|
||||
|
||||
// ErrRunFinished is returned when a redelivered start message names a run that
|
||||
// already reached a terminal state. The run must not be resurrected; callers
|
||||
// should log and ack instead of retrying.
|
||||
var ErrRunFinished = fmt.Errorf("monthly usage run already finished")
|
||||
|
||||
// runAlreadyFinished reports whether runID exists and has already completed.
|
||||
// A missing run is not an error: the message may be the first delivery.
|
||||
func (p *Producer) runAlreadyFinished(ctx context.Context, runID string) (bool, error) {
|
||||
meta, err := p.store.GetRunMeta(ctx, runID)
|
||||
if err != nil {
|
||||
if strings.Contains(err.Error(), "not found") {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
return meta.Status == "completed" || meta.Status == "completed_with_failures", nil
|
||||
}
|
||||
|
||||
func (p *Producer) Produce(ctx context.Context, message model.TrafficUsageMonthlyStartMessage) (err error) {
|
||||
startedAt := time.Now()
|
||||
stats := productionStats{channels: make(map[string]struct{})}
|
||||
if !p.running.CompareAndSwap(0, 1) {
|
||||
stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "precheck", "already_running", startedAt, 0, nil)
|
||||
return ErrAlreadyRunning
|
||||
}
|
||||
defer p.running.Store(0)
|
||||
|
||||
if message.RunID == "" {
|
||||
err = fmt.Errorf("runId is empty")
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, "validate", "", startedAt, 0, err)
|
||||
return err
|
||||
}
|
||||
scope, err := model.NormalizeScope(message.Scope)
|
||||
if err != nil {
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, "validate", "invalid_scope", startedAt, 0, err)
|
||||
return err
|
||||
}
|
||||
stats.scopeChannels = scope.ChannelCount()
|
||||
// A redelivered start message must not resurrect a run that already
|
||||
// finished; that would create a second producer pass over the same RunID.
|
||||
if p.config.RoundRetryEnabled {
|
||||
if finished, checkErr := p.runAlreadyFinished(ctx, message.RunID); checkErr != nil {
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, "run_state", "", startedAt, 0, checkErr)
|
||||
return checkErr
|
||||
} else if finished {
|
||||
stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "run_state", "run_finished", startedAt, 0, nil)
|
||||
return ErrRunFinished
|
||||
}
|
||||
}
|
||||
acquired, err := p.store.AcquireLockOrTakeover(ctx, p.config.RunLockKey, p.owner, message.RunID, p.config.RunLockTTL)
|
||||
if err != nil {
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, "lock", "", startedAt, 0, err)
|
||||
return err
|
||||
}
|
||||
if !acquired {
|
||||
stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "lock", "already_running", startedAt, 0, nil)
|
||||
return ErrAlreadyRunning
|
||||
}
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
defer func() {
|
||||
releaseCtx, releaseCancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer releaseCancel()
|
||||
_, _ = p.store.ReleaseLock(releaseCtx, p.config.RunLockKey, p.owner)
|
||||
}()
|
||||
pending, err := p.hasPendingTasks(ctx, message.RunID)
|
||||
if err != nil {
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, "pending_check", "", startedAt, 0, err)
|
||||
return err
|
||||
}
|
||||
if pending {
|
||||
stats.log(ctx, "skipped", message.RunID, message.QueryMonth, "pending_check", "pending_tasks", startedAt, 0, nil)
|
||||
return ErrAlreadyRunning
|
||||
}
|
||||
renewErr := make(chan error, 1)
|
||||
renewDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(renewDone)
|
||||
p.renewLock(runCtx, cancel, renewErr)
|
||||
}()
|
||||
defer func() {
|
||||
cancel()
|
||||
<-renewDone
|
||||
select {
|
||||
case lockErr := <-renewErr:
|
||||
err = lockErr
|
||||
default:
|
||||
}
|
||||
}()
|
||||
ctx = runCtx
|
||||
if message.Operation == "" {
|
||||
message.Operation = operation
|
||||
}
|
||||
if p.config.RoundRetryEnabled {
|
||||
if err = p.store.CreateRun(ctx, zset.RunMeta{
|
||||
RunID: message.RunID,
|
||||
Operation: message.Operation,
|
||||
QueryMonth: message.QueryMonth,
|
||||
Status: "producing",
|
||||
CurrentRound: 1,
|
||||
ScopeChannels: scope.ChannelCount(),
|
||||
ScopeDigest: scopeDigest(scope),
|
||||
ScopeChannelsSample: strings.Join(scope.ChannelSample(scopeSampleLimit), ","),
|
||||
}); err != nil {
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, "run_init", "", startedAt, 0, err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
maxID, err := p.scanner.SnapshotMaxID(ctx)
|
||||
if err != nil {
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, "snapshot", "", startedAt, 0, err)
|
||||
return err
|
||||
}
|
||||
g.Log().Infof(ctx, "event=monthly_usage_producer status=started runId=%s queryMonth=%s scanUpperID=%d scopeChannels=%d scanMode=channel scanConcurrency=%d pageSize=%d", message.RunID, message.QueryMonth, maxID, scope.ChannelCount(), p.config.ScanConcurrency, p.config.PageSize)
|
||||
stage := "scan"
|
||||
defer func() {
|
||||
if err != nil {
|
||||
stats.log(ctx, "failed", message.RunID, message.QueryMonth, stage, "", startedAt, maxID, err)
|
||||
return
|
||||
}
|
||||
stats.log(ctx, "completed", message.RunID, message.QueryMonth, "", "", startedAt, maxID, nil)
|
||||
}()
|
||||
if p.config.RoundRetryEnabled {
|
||||
stage = "channel_scan"
|
||||
if err = p.produceChannels(ctx, message, scope, maxID, startedAt, &stats); err != nil {
|
||||
if isProductionClosed(err) {
|
||||
// A same-Run redelivery after producerDone leaves consumption alone.
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
// Verify ownership immediately before closing production. The periodic
|
||||
// renewer still cancels scanners and capacity waits on lease loss.
|
||||
var owned bool
|
||||
owned, err = p.store.RenewLock(ctx, p.config.RunLockKey, p.owner, p.config.RunLockTTL)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !owned {
|
||||
return fmt.Errorf("producer lock token is no longer owned")
|
||||
}
|
||||
if err = ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
stage = "producer_done"
|
||||
err = p.store.MarkProducerDone(ctx, message.RunID)
|
||||
return err
|
||||
}
|
||||
return fmt.Errorf("monthly usage full sync requires Run queue; legacy serial production has been removed")
|
||||
}
|
||||
|
||||
// isRunNotFound reports whether err means the run's metadata hash is absent, as
|
||||
// opposed to a genuine read failure.
|
||||
func isRunNotFound(err error) bool {
|
||||
return errors.Is(err, zset.ErrRunNotFound)
|
||||
}
|
||||
|
||||
func (p *Producer) hasPendingTasks(ctx context.Context, currentRunID ...string) (bool, error) {
|
||||
currentID := ""
|
||||
if len(currentRunID) > 0 {
|
||||
currentID = currentRunID[0]
|
||||
}
|
||||
if p.config.RoundRetryEnabled {
|
||||
runs, err := p.store.ActiveRuns(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
for _, runID := range runs {
|
||||
if runID == currentID {
|
||||
// A redelivered message for the same run may resume a producer
|
||||
// that stopped before producerDone. Stable task IDs make the
|
||||
// rescan idempotent; other runs still provide backpressure.
|
||||
continue
|
||||
}
|
||||
meta, metaErr := p.store.GetRunMeta(ctx, runID)
|
||||
if metaErr != nil {
|
||||
// The active index is a global key written outside the run-scoped
|
||||
// Lua transaction, so it can list a run whose hash is already gone
|
||||
// (expired, evicted, or cleaned by hand). Returning an error here
|
||||
// would surface as a non-retryable failure and make the consumer
|
||||
// redeliver this request forever. Treat it as a stale index entry
|
||||
// and drop it.
|
||||
if isRunNotFound(metaErr) {
|
||||
if _, removeErr := p.store.RedisClient().Do(ctx, "SREM", p.store.ActiveRunsKey(), runID); removeErr != nil {
|
||||
return false, removeErr
|
||||
}
|
||||
g.Log().Warningf(ctx, "event=monthly_usage_producer status=skipped stage=pending_check reason=stale_index runId=%s", runID)
|
||||
continue
|
||||
}
|
||||
return false, metaErr
|
||||
}
|
||||
// A "running" run has already finished producing (producerDone=1)
|
||||
// and the consumer owns the remaining work: it is genuinely in-flight
|
||||
// and must keep providing backpressure.
|
||||
if meta.Status == "running" {
|
||||
return true, nil
|
||||
}
|
||||
// A "producing" run for another ID cannot have a live producer. This
|
||||
// function only runs after AcquireLockOrTakeover succeeded on the
|
||||
// single producer lock, so any live producer would still hold it and
|
||||
// we could not have acquired it. Such a run is a zombie left behind by
|
||||
// a producer that died before MarkProducerDone (for example the
|
||||
// process was replaced mid-run).
|
||||
//
|
||||
// Promote it so the normal lifecycle can drain and close it. Blocking
|
||||
// would reject every later request forever, and merely warning would
|
||||
// leak the run's keys and its active-index entry indefinitely.
|
||||
reclaimed, reclaimErr := p.store.MarkProducerAbandoned(ctx, runID)
|
||||
if reclaimErr != nil {
|
||||
return false, reclaimErr
|
||||
}
|
||||
g.Log().Warningf(ctx, "event=monthly_usage_producer status=skipped stage=pending_check reason=zombie_run runId=%s reclaimed=%t activeCount=%d inflightCount=%d nextRoundCount=%d", runID, reclaimed, meta.ActiveCount, meta.InflightCount, meta.NextRoundCount)
|
||||
}
|
||||
}
|
||||
channels, err := p.store.ActiveChannels(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
for _, chanKey := range channels {
|
||||
size, sizeErr := p.store.Size(ctx, chanKey)
|
||||
if sizeErr != nil {
|
||||
return false, sizeErr
|
||||
}
|
||||
if size > 0 {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
return false, nil
|
||||
}
|
||||
|
||||
func (p *Producer) renewLock(ctx context.Context, cancel context.CancelFunc, errors chan<- error) {
|
||||
interval := p.config.RunLockTTL / 3
|
||||
if interval <= 0 {
|
||||
interval = time.Second
|
||||
}
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
ok, err := p.store.RenewLock(ctx, p.config.RunLockKey, p.owner, p.config.RunLockTTL)
|
||||
if err != nil || !ok {
|
||||
// Cancellation during a renewal RPC is shutdown, not lease loss.
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if err == nil {
|
||||
err = fmt.Errorf("producer lock token is no longer owned")
|
||||
}
|
||||
select {
|
||||
case errors <- fmt.Errorf("renew producer lock: %w", err):
|
||||
default:
|
||||
}
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func newOwner() string {
|
||||
var bytes [12]byte
|
||||
if _, err := rand.Read(bytes[:]); err == nil {
|
||||
return "producer-" + hex.EncodeToString(bytes[:])
|
||||
}
|
||||
return fmt.Sprintf("producer-%d", time.Now().UnixNano())
|
||||
}
|
||||
|
||||
func sleepContext(ctx context.Context, duration time.Duration) error {
|
||||
timer := time.NewTimer(duration)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-timer.C:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/gogf/gf/v2/container/gvar"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
||||
)
|
||||
|
||||
type pendingRedis struct {
|
||||
channels []string
|
||||
sizes map[string]int64
|
||||
failKey string
|
||||
}
|
||||
|
||||
func (r *pendingRedis) Do(_ context.Context, command string, args ...any) (*gvar.Var, error) {
|
||||
switch strings.ToUpper(command) {
|
||||
case "SMEMBERS":
|
||||
return gvar.New(r.channels), nil
|
||||
case "ZCARD":
|
||||
key := fmt.Sprint(args[0])
|
||||
if key == r.failKey {
|
||||
return nil, fmt.Errorf("redis unavailable")
|
||||
}
|
||||
return gvar.New(r.sizes[key]), nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unexpected command %s", command)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *pendingRedis) Eval(_ context.Context, script string, numKeys int64, keys []string, args []any) (*gvar.Var, error) {
|
||||
return nil, fmt.Errorf("unexpected Eval call")
|
||||
}
|
||||
|
||||
func TestHasPendingTasks(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
redis *pendingRedis
|
||||
want bool
|
||||
wantErr bool
|
||||
}{
|
||||
{
|
||||
name: "all registered channels are empty",
|
||||
redis: &pendingRedis{
|
||||
channels: []string{"mobile-a", "telecom-a"},
|
||||
sizes: map[string]int64{
|
||||
"flow:mobile-a": 0,
|
||||
"flow:telecom-a": 0,
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "previous run still has tasks",
|
||||
redis: &pendingRedis{
|
||||
channels: []string{"mobile-a", "telecom-a"},
|
||||
sizes: map[string]int64{
|
||||
"flow:mobile-a": 0,
|
||||
"flow:telecom-a": 3,
|
||||
},
|
||||
},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "redis error is not treated as empty",
|
||||
redis: &pendingRedis{
|
||||
channels: []string{"mobile-a"},
|
||||
sizes: map[string]int64{},
|
||||
failKey: "flow:mobile-a",
|
||||
},
|
||||
wantErr: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
store, err := zset.New(test.redis, "flow:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
producer, err := New(&fakeScanner{}, store, Config{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got, err := func() (bool, error) {
|
||||
producer.config.RoundRetryEnabled = false
|
||||
return producer.hasPendingTasks(context.Background())
|
||||
}()
|
||||
if (err != nil) != test.wantErr {
|
||||
t.Fatalf("hasPendingTasks() error = %v, wantErr %v", err, test.wantErr)
|
||||
}
|
||||
if got != test.want {
|
||||
t.Fatalf("hasPendingTasks() = %v, want %v", got, test.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type fakeScanner struct{}
|
||||
|
||||
func (*fakeScanner) SnapshotMaxID(context.Context) (int64, error) { return 0, nil }
|
||||
func (*fakeScanner) Scan(context.Context, int64, int64, int) ([]Card, error) {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func (*fakeScanner) ListChannels(context.Context) ([]channelMetadata, error) { return nil, nil }
|
||||
func (*fakeScanner) ScanChannel(context.Context, string, int64, int) ([]Card, error) { return nil, nil }
|
||||
|
||||
type pageScanner struct {
|
||||
cards []Card
|
||||
}
|
||||
|
||||
func (s *pageScanner) SnapshotMaxID(context.Context) (int64, error) {
|
||||
if len(s.cards) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
return s.cards[len(s.cards)-1].ID, nil
|
||||
}
|
||||
|
||||
func (s *pageScanner) Scan(_ context.Context, lastID, _ int64, _ int) ([]Card, error) {
|
||||
if lastID != 0 {
|
||||
return nil, nil
|
||||
}
|
||||
return s.cards, nil
|
||||
}
|
||||
|
||||
type producerRedis struct {
|
||||
mu sync.Mutex
|
||||
enqueueBatches map[string]int
|
||||
}
|
||||
|
||||
func (r *producerRedis) Do(_ context.Context, command string, _ ...any) (*gvar.Var, error) {
|
||||
switch strings.ToUpper(command) {
|
||||
case "SET":
|
||||
return gvar.New("OK"), nil
|
||||
case "SMEMBERS":
|
||||
return gvar.New([]string{}), nil
|
||||
case "HGETALL":
|
||||
// No stored run metadata: the producer treats this as a first delivery.
|
||||
return gvar.New(map[string]string{}), nil
|
||||
default:
|
||||
return nil, fmt.Errorf("unexpected command %s", command)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *producerRedis) Eval(_ context.Context, _ string, numKeys int64, keys []string, args []any) (*gvar.Var, error) {
|
||||
if numKeys == 1 {
|
||||
return gvar.New(int64(1)), nil
|
||||
}
|
||||
if numKeys != 3 {
|
||||
return nil, fmt.Errorf("unexpected key count %d", numKeys)
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.enqueueBatches[keys[0]] = len(args) - 3
|
||||
r.mu.Unlock()
|
||||
return gvar.New(len(args) - 3), nil
|
||||
}
|
||||
|
||||
func (s *pageScanner) ListChannels(ctx context.Context) ([]channelMetadata, error) {
|
||||
return scannerFor(s.cards, 0).ListChannels(ctx)
|
||||
}
|
||||
func (s *pageScanner) ScanChannel(ctx context.Context, key string, last int64, size int) ([]Card, error) {
|
||||
return scannerFor(s.cards, 0).ScanChannel(ctx, key, last, size)
|
||||
}
|
||||
@@ -0,0 +1,249 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
||||
)
|
||||
|
||||
func newScopeProducer(t *testing.T, cards []Card) (*Producer, *producerRedis) {
|
||||
t.Helper()
|
||||
store := testStore(t)
|
||||
scanner := scannerFor(cards, cards[len(cards)-1].ID)
|
||||
for i := range scanner.channels {
|
||||
for _, c := range cards {
|
||||
if c.ChanKey == scanner.channels[i].ChanKey {
|
||||
scanner.channels[i].TempKey = c.TempKey
|
||||
scanner.channels[i].SyncEnabled = c.SyncEnabled
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
p, err := New(scanner, store, Config{PageSize: 100})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p, &producerRedis{enqueueBatches: map[string]int{}}
|
||||
}
|
||||
func scopedMessage(runID string, channels ...string) model.TrafficUsageMonthlyStartMessage {
|
||||
return model.TrafficUsageMonthlyStartMessage{
|
||||
RunID: runID,
|
||||
Operation: "current_month_usage_full_sync",
|
||||
QueryMonth: "202608",
|
||||
Scope: &model.SyncScope{Channels: channels},
|
||||
}
|
||||
}
|
||||
|
||||
func TestProduceScopeEnqueuesOnlySelectedChannels(t *testing.T) {
|
||||
producer, redis := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", Operator: 1},
|
||||
{ID: 3, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
{ID: 4, ChanKey: "189", TempKey: "sim_yd_v5", Operator: 1},
|
||||
})
|
||||
|
||||
if err := producer.Produce(context.Background(), scopedMessage("scoped", "217")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
redis.enqueueBatches = map[string]int{}
|
||||
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
||||
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
||||
n, _ := producer.store.SizeKey(context.Background(), key)
|
||||
if n > 0 {
|
||||
redis.enqueueBatches["flow:"+ch] = int(n)
|
||||
}
|
||||
}
|
||||
redis.mu.Lock()
|
||||
defer redis.mu.Unlock()
|
||||
if len(redis.enqueueBatches) != 1 {
|
||||
t.Fatalf("enqueue batches = %+v, want only the 217 channel", redis.enqueueBatches)
|
||||
}
|
||||
if redis.enqueueBatches["flow:217"] != 2 {
|
||||
t.Fatalf("enqueue batches = %+v, want 2 tasks on flow:217", redis.enqueueBatches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProduceScopeMultipleChannels(t *testing.T) {
|
||||
producer, redis := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", Operator: 1},
|
||||
{ID: 3, ChanKey: "189", TempKey: "sim_yd_v5", Operator: 1},
|
||||
})
|
||||
|
||||
if err := producer.Produce(context.Background(), scopedMessage("scoped", "189", "217")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
redis.enqueueBatches = map[string]int{}
|
||||
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
||||
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
||||
n, _ := producer.store.SizeKey(context.Background(), key)
|
||||
if n > 0 {
|
||||
redis.enqueueBatches["flow:"+ch] = int(n)
|
||||
}
|
||||
}
|
||||
redis.mu.Lock()
|
||||
defer redis.mu.Unlock()
|
||||
if len(redis.enqueueBatches) != 2 {
|
||||
t.Fatalf("enqueue batches = %+v, want 217 and 189", redis.enqueueBatches)
|
||||
}
|
||||
if _, ok := redis.enqueueBatches["flow:173"]; ok {
|
||||
t.Fatalf("channel 173 was enqueued despite not being selected: %+v", redis.enqueueBatches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProduceFullScopeKeepsEveryChannel(t *testing.T) {
|
||||
producer, redis := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", Operator: 1},
|
||||
})
|
||||
|
||||
if err := producer.Produce(context.Background(), model.TrafficUsageMonthlyStartMessage{
|
||||
RunID: "full",
|
||||
Operation: "current_month_usage_full_sync",
|
||||
QueryMonth: "202608",
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
redis.enqueueBatches = map[string]int{}
|
||||
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
||||
key, _ := producer.store.RoundActiveKey("full", 1, ch)
|
||||
n, _ := producer.store.SizeKey(context.Background(), key)
|
||||
if n > 0 {
|
||||
redis.enqueueBatches["flow:"+ch] = int(n)
|
||||
}
|
||||
}
|
||||
redis.mu.Lock()
|
||||
defer redis.mu.Unlock()
|
||||
if len(redis.enqueueBatches) != 2 {
|
||||
t.Fatalf("enqueue batches = %+v, want both channels for a full sync", redis.enqueueBatches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProduceScopeCombinesWithSyncDisabledGuard(t *testing.T) {
|
||||
producer, redis := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
// Selected but excluded by sync_enabled: scope must not override it.
|
||||
{ID: 2, ChanKey: "173", TempKey: "sim_yd_v5", SyncEnabled: true, Operator: 1},
|
||||
})
|
||||
|
||||
if err := producer.Produce(context.Background(), scopedMessage("scoped", "173")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
redis.enqueueBatches = map[string]int{}
|
||||
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
||||
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
||||
n, _ := producer.store.SizeKey(context.Background(), key)
|
||||
if n > 0 {
|
||||
redis.enqueueBatches["flow:"+ch] = int(n)
|
||||
}
|
||||
}
|
||||
redis.mu.Lock()
|
||||
defer redis.mu.Unlock()
|
||||
if len(redis.enqueueBatches) != 0 {
|
||||
t.Fatalf("enqueue batches = %+v, want none (selected channel is sync-disabled)", redis.enqueueBatches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProduceScopeSelectingUnknownChannelEnqueuesNothing(t *testing.T) {
|
||||
producer, redis := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
})
|
||||
|
||||
if err := producer.Produce(context.Background(), scopedMessage("scoped", "does-not-exist")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
redis.enqueueBatches = map[string]int{}
|
||||
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
||||
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
||||
n, _ := producer.store.SizeKey(context.Background(), key)
|
||||
if n > 0 {
|
||||
redis.enqueueBatches["flow:"+ch] = int(n)
|
||||
}
|
||||
}
|
||||
redis.mu.Lock()
|
||||
defer redis.mu.Unlock()
|
||||
if len(redis.enqueueBatches) != 0 {
|
||||
t.Fatalf("enqueue batches = %+v, want none", redis.enqueueBatches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProduceRejectsInvalidScope(t *testing.T) {
|
||||
producer, redis := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
})
|
||||
|
||||
if err := producer.Produce(context.Background(), scopedMessage("bad", "invalid/chan")); err == nil {
|
||||
t.Fatal("expected invalid chanKey to be rejected")
|
||||
}
|
||||
redis.enqueueBatches = map[string]int{}
|
||||
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
||||
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
||||
n, _ := producer.store.SizeKey(context.Background(), key)
|
||||
if n > 0 {
|
||||
redis.enqueueBatches["flow:"+ch] = int(n)
|
||||
}
|
||||
}
|
||||
redis.mu.Lock()
|
||||
defer redis.mu.Unlock()
|
||||
if len(redis.enqueueBatches) != 0 {
|
||||
t.Fatalf("rejected scope must not enqueue: %+v", redis.enqueueBatches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProduceScopeDeduplicatesChannels(t *testing.T) {
|
||||
producer, redis := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
})
|
||||
|
||||
if err := producer.Produce(context.Background(), scopedMessage("scoped", "217", "217")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
redis.enqueueBatches = map[string]int{}
|
||||
for _, ch := range []string{"217", "173", "189", "ghost-channel", "a", "b", "mobile-a", "mobile-b"} {
|
||||
key, _ := producer.store.RoundActiveKey("scoped", 1, ch)
|
||||
n, _ := producer.store.SizeKey(context.Background(), key)
|
||||
if n > 0 {
|
||||
redis.enqueueBatches["flow:"+ch] = int(n)
|
||||
}
|
||||
}
|
||||
redis.mu.Lock()
|
||||
defer redis.mu.Unlock()
|
||||
if redis.enqueueBatches["flow:217"] != 1 {
|
||||
t.Fatalf("enqueue batches = %+v, want 1 task despite a duplicated chanKey", redis.enqueueBatches)
|
||||
}
|
||||
}
|
||||
|
||||
func TestScopeDigestIsOrderIndependent(t *testing.T) {
|
||||
first := &model.SyncScope{Channels: []string{"173", "217"}}
|
||||
second := &model.SyncScope{Channels: []string{"217", "173"}}
|
||||
// Normalization sorts the channels, so both requests produce one digest.
|
||||
normalizedFirst, err := model.NormalizeScope(first)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
normalizedSecond, err := model.NormalizeScope(second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if scopeDigest(normalizedFirst) != scopeDigest(normalizedSecond) {
|
||||
t.Fatalf("digest depends on channel order: %s vs %s", scopeDigest(normalizedFirst), scopeDigest(normalizedSecond))
|
||||
}
|
||||
if scopeDigest(nil) != "" {
|
||||
t.Fatalf("full sync must have an empty digest")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunAlreadyFinishedTreatsMissingRunAsFirstDelivery(t *testing.T) {
|
||||
producer, _ := newScopeProducer(t, []Card{
|
||||
{ID: 1, ChanKey: "217", TempKey: "sim_yd_v5", Operator: 1},
|
||||
})
|
||||
// The fake Redis returns no metadata, which must not be treated as finished.
|
||||
finished, err := producer.runAlreadyFinished(context.Background(), "fresh")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %v", err)
|
||||
}
|
||||
if finished {
|
||||
t.Fatal("a run without stored metadata must not be reported as finished")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/supply"
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"time"
|
||||
)
|
||||
|
||||
func (p *Producer) observe(key, stage string, pending int, added, ended bool) {
|
||||
if p.observation != nil {
|
||||
p.observation.Event(key, stage, pending, added, ended, time.Now())
|
||||
}
|
||||
}
|
||||
func (p *Producer) startObservation(ctx context.Context, run string, keys []string, started time.Time) func(bool) {
|
||||
tracker := p.config.SupplyObservation
|
||||
if tracker == nil {
|
||||
return func(bool) {}
|
||||
}
|
||||
o := supply.New(newOwner(), started, keys)
|
||||
p.observation = o
|
||||
readCtx, readCancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
meta, err := p.store.GetRunMeta(readCtx, run)
|
||||
readCancel()
|
||||
if err != nil {
|
||||
o.Partial()
|
||||
} else if meta.SupplyObservation != nil {
|
||||
o.Restore(*meta.SupplyObservation)
|
||||
}
|
||||
tracker.Put(run, o)
|
||||
stop := make(chan struct{})
|
||||
done := make(chan struct{})
|
||||
flush := func() {
|
||||
writeCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
defer cancel()
|
||||
encoded, e := o.Encode()
|
||||
if e == nil {
|
||||
var saved bool
|
||||
saved, e = p.store.SaveSupply(writeCtx, run, p.config.RunLockKey, p.owner, encoded)
|
||||
if !saved && e == nil {
|
||||
o.Partial()
|
||||
}
|
||||
}
|
||||
if e != nil {
|
||||
o.Partial()
|
||||
g.Log().Warningf(writeCtx, "event=supply_observation status=partial runId=%s error=%q", run, e.Error())
|
||||
}
|
||||
}
|
||||
go func() {
|
||||
defer close(done)
|
||||
sampleTicker := time.NewTicker(time.Second)
|
||||
defer sampleTicker.Stop()
|
||||
flushTicker := time.NewTicker(10 * time.Second)
|
||||
defer flushTicker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
case <-ctx.Done():
|
||||
o.Partial()
|
||||
return
|
||||
case <-flushTicker.C:
|
||||
flush()
|
||||
case <-sampleTicker.C:
|
||||
views := o.Views()
|
||||
sampleCtx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
samples, e := p.store.ReadSupply(sampleCtx, run, keys, time.Now())
|
||||
cancel()
|
||||
now := time.Now()
|
||||
for _, key := range keys {
|
||||
s := samples[key]
|
||||
if e != nil || !tracker.LocalConsumer || !tracker.ConsumerActive.Load() {
|
||||
s.Known = false
|
||||
}
|
||||
o.Apply(key, views[key], s, now)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return func(success bool) {
|
||||
close(stop)
|
||||
<-done
|
||||
if !success {
|
||||
o.Partial()
|
||||
}
|
||||
flush()
|
||||
tracker.Remove(run)
|
||||
p.observation = nil
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/fullsync/supply"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/testutil"
|
||||
"github.com/gogf/gf/v2/container/gvar"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type telemetryFailRedis struct{ zset.RedisClient }
|
||||
|
||||
func (r telemetryFailRedis) Eval(ctx context.Context, script string, n int64, keys []string, args []any) (*gvar.Var, error) {
|
||||
if strings.Contains(script, "'supplyObservation'") {
|
||||
return nil, fmt.Errorf("injected telemetry write failure")
|
||||
}
|
||||
return r.RedisClient.Eval(ctx, script, n, keys, args)
|
||||
}
|
||||
func TestSupplyFlushFailureDoesNotBlockProducerDone(t *testing.T) {
|
||||
client := testutil.Redis(t)
|
||||
store, err := zset.New(telemetryFailRedis{client}, "flow:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}}, 1)
|
||||
p := testProducer(t, scanner, store, Config{SupplyObservation: supply.NewTracker(true)})
|
||||
if err = p.Produce(context.Background(), model.TrafficUsageMonthlyStartMessage{RunID: "telemetry-failure"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
requireMeta(t, store, "telemetry-failure", true, 1)
|
||||
}
|
||||
|
||||
func TestSupplyObservesEmptyTargetAndFinalFlush(t *testing.T) {
|
||||
store := testStore(t)
|
||||
scanner := scannerFor([]Card{{ID: 1, ChanKey: "a"}}, 2)
|
||||
scanner.channels = append(scanner.channels, channelMetadata{ChanKey: "empty", TempKey: "sim_yd_v5"})
|
||||
tracker := supply.NewTracker(true)
|
||||
tracker.ConsumerActive.Store(true)
|
||||
scanner.hook = func(ctx context.Context, key string, last int64) error {
|
||||
if key == "empty" {
|
||||
return sleepContext(ctx, 2200*time.Millisecond)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
p := testProducer(t, scanner, store, Config{SupplyObservation: tracker})
|
||||
if err := p.Produce(context.Background(), model.TrafficUsageMonthlyStartMessage{RunID: "supply-end", QueryMonth: "202610"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, err := store.GetRunMeta(context.Background(), "supply-end")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !m.ProducerDone || m.SupplyObservation == nil {
|
||||
t.Fatalf("missing final observation %+v", m)
|
||||
}
|
||||
snapshot := m.SupplyObservation
|
||||
if snapshot.Summary.Targets != 2 || snapshot.Summary.Confirmed != 0 {
|
||||
t.Fatalf("bad summary %+v", snapshot)
|
||||
}
|
||||
c := snapshot.Channels[1]
|
||||
if c.Key != "empty" || c.FirstEnqueueMS != nil || c.Termination != "empty_result" || c.Unconfirmed.TotalMS < 900 {
|
||||
t.Fatalf("bad empty target %+v", c)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
package monthlyusage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/queue/zset"
|
||||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/testutil"
|
||||
)
|
||||
|
||||
func newZombieTestProducer(t *testing.T) (*Producer, *zset.Store) {
|
||||
t.Helper()
|
||||
store, err := zset.New(testutil.Redis(t), "flow:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
producer, err := New(&pageScanner{cards: nil}, store, Config{PageSize: 100, RoundRetryEnabled: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return producer, store
|
||||
}
|
||||
|
||||
// TestHasPendingTasksReclaimsZombieRun guards the "zombie run" regression.
|
||||
//
|
||||
// A producer that dies before MarkProducerDone (for example the container is
|
||||
// replaced mid-run) leaves its run in status=producing with producerDone=0.
|
||||
// completeRunIfIdle and advanceRunRound both require producerDone=1, so such a
|
||||
// run can never be closed, never leaves the active index, and never gets the
|
||||
// retention TTL. Treating it as backpressure rejected every later request
|
||||
// forever; merely warning about it leaked keys and an index entry indefinitely.
|
||||
func TestHasPendingTasksReclaimsZombieRun(t *testing.T) {
|
||||
producer, store := newZombieTestProducer(t)
|
||||
ctx := context.Background()
|
||||
|
||||
// A genuinely in-flight run (producerDone set, consumer working) must keep
|
||||
// providing backpressure.
|
||||
if err := store.CreateRun(ctx, zset.RunMeta{RunID: "running-run", Status: "running", ProducerDone: true, CurrentRound: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pending, err := producer.hasPendingTasks(ctx, "new-run")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !pending {
|
||||
t.Fatal("a running run must still provide backpressure")
|
||||
}
|
||||
if _, err := store.RedisClient().Do(ctx, "SREM", store.ActiveRunsKey(), "running-run"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// A zombie run must not block a new request, and must be promoted so the
|
||||
// normal lifecycle can drain and close it.
|
||||
if err := store.CreateRun(ctx, zset.RunMeta{RunID: "zombie-run", Status: "producing", ProducerDone: false, CurrentRound: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pending, err = producer.hasPendingTasks(ctx, "new-run-2")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if pending {
|
||||
t.Fatal("a producing run with producerDone=0 must not block new requests (zombie run)")
|
||||
}
|
||||
|
||||
meta, err := store.GetRunMeta(ctx, "zombie-run")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !meta.ProducerDone {
|
||||
t.Fatal("the zombie run must be promoted (producerDone=1) so it can be closed")
|
||||
}
|
||||
if meta.Status != "running" {
|
||||
t.Fatalf("promoted run status = %q, want running", meta.Status)
|
||||
}
|
||||
// Promotion makes the run closable, which is what applies the retention TTL
|
||||
// and releases the active index entry.
|
||||
if completed, err := store.CompleteRunIfIdle(ctx, "zombie-run"); err != nil || !completed {
|
||||
t.Fatalf("promoted run must be closable: completed=%v err=%v", completed, err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHasPendingTasksNeverReclaimsReplayRun guards the boundary: a replay target
|
||||
// is also created as "producing", but it is driven by the replay engine under
|
||||
// its own lock, so its producerDone is not the producer's to set.
|
||||
func TestHasPendingTasksNeverReclaimsReplayRun(t *testing.T) {
|
||||
producer, store := newZombieTestProducer(t)
|
||||
ctx := context.Background()
|
||||
|
||||
if err := store.CreateRun(ctx, zset.RunMeta{
|
||||
RunID: "replay-run", Status: "producing", ProducerDone: false, CurrentRound: 1,
|
||||
SyncType: "replay", SourceRunID: "source-run", ReplayCycle: 1,
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := producer.hasPendingTasks(ctx, "new-run"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
meta, err := store.GetRunMeta(ctx, "replay-run")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if meta.ProducerDone {
|
||||
t.Fatal("a replay run must not be promoted by the full-sync producer")
|
||||
}
|
||||
if meta.Status != "producing" {
|
||||
t.Fatalf("replay run status = %q, want producing", meta.Status)
|
||||
}
|
||||
}
|
||||
|
||||
// TestHasPendingTasksDropsStaleIndexEntry covers a second unrecoverable path:
|
||||
// the active index can list a run whose hash is already gone. Returning that as
|
||||
// an error made the request fail permanently (the consumer redelivers it
|
||||
// forever), which is the same "only fixable by hand-cleaning Redis" failure.
|
||||
func TestHasPendingTasksDropsStaleIndexEntry(t *testing.T) {
|
||||
producer, store := newZombieTestProducer(t)
|
||||
ctx := context.Background()
|
||||
|
||||
// Index a run that has no metadata hash.
|
||||
if _, err := store.RedisClient().Do(ctx, "SADD", store.ActiveRunsKey(), "ghost-run"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pending, err := producer.hasPendingTasks(ctx, "new-run")
|
||||
if err != nil {
|
||||
t.Fatalf("a stale index entry must not fail the request: %v", err)
|
||||
}
|
||||
if pending {
|
||||
t.Fatal("a stale index entry must not provide backpressure")
|
||||
}
|
||||
// The stale entry is removed so it does not have to be handled again.
|
||||
members, err := store.ActiveRuns(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, member := range members {
|
||||
if member == "ghost-run" {
|
||||
t.Fatal("the stale index entry must be removed")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestProduceCreatesRunAfterZombie covers the end-to-end path through Produce:
|
||||
// a zombie run must not stop a new run from being created.
|
||||
func TestProduceCreatesRunAfterZombie(t *testing.T) {
|
||||
store, err := zset.New(testutil.Redis(t), "flow:")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
producer, err := New(&pageScanner{cards: []Card{
|
||||
{ID: 1, ChanKey: "channel-a", TempKey: "sim_yd_v5", Operator: 1},
|
||||
}}, store, Config{PageSize: 100, RoundRetryEnabled: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// A zombie run from a crashed producer.
|
||||
if err := store.CreateRun(ctx, zset.RunMeta{RunID: "zombie-run", Status: "producing", ProducerDone: false, CurrentRound: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// A new request must still produce its own run.
|
||||
if err := producer.Produce(ctx, scopedMessage("fresh-run", "channel-a")); err != nil {
|
||||
t.Fatalf("a zombie run must not stop a new run: %v", err)
|
||||
}
|
||||
meta, err := store.GetRunMeta(ctx, "fresh-run")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The producer enqueues into the active ZSet; ActiveCount is the count it
|
||||
// maintains. EnqueuedCount is an accounting field the producer does not set.
|
||||
if !meta.ProducerDone || meta.ActiveCount != 1 {
|
||||
t.Fatalf("fresh run metadata = %+v, want producerDone with 1 active task", meta)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user