diff --git a/pkg/costcalc/metrics/metrics.go b/pkg/costcalc/metrics/metrics.go index 0e03b0db..80a83299 100644 --- a/pkg/costcalc/metrics/metrics.go +++ b/pkg/costcalc/metrics/metrics.go @@ -57,6 +57,13 @@ var ( Help: "Cost results that superseded a previous version.", }) + // UpsertRetryTotal increments when UpsertWithSupersede retries after a + // concurrent insert caused a uk_cpc_active unique violation. + UpsertRetryTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "finance_cost_upsert_retry_total", + Help: "UpsertWithSupersede retries due to concurrent unique violation.", + }) + // AuditWritesTotal increments per aud_cost_history row written. AuditWritesTotal = promauto.NewCounter(prometheus.CounterOpts{ Name: "finance_cost_audit_writes_total", diff --git a/services/finance-cost-orchestrator/internal/config/config.go b/services/finance-cost-orchestrator/internal/config/config.go index 832c019b..4502114d 100644 --- a/services/finance-cost-orchestrator/internal/config/config.go +++ b/services/finance-cost-orchestrator/internal/config/config.go @@ -65,10 +65,12 @@ type RabbitMQConfig struct { // OrchestratorConfig holds orchestrator-specific tuning knobs. type OrchestratorConfig struct { - ChunkSize int `mapstructure:"chunk_size"` - MaxChunkSize int `mapstructure:"max_chunk_size"` - CronSchedule string `mapstructure:"cron_schedule"` - CronTimezone string `mapstructure:"cron_timezone"` + ChunkSize int `mapstructure:"chunk_size"` + MaxChunkSize int `mapstructure:"max_chunk_size"` + CronSchedule string `mapstructure:"cron_schedule"` + CronTimezone string `mapstructure:"cron_timezone"` + StuckChunkTimeout time.Duration `mapstructure:"stuck_chunk_timeout"` + StuckSweepInterval time.Duration `mapstructure:"stuck_sweep_interval"` } // TracingConfig holds Jaeger/OpenTelemetry configuration. @@ -144,6 +146,8 @@ func setDefaults(v *viper.Viper) { v.SetDefault("orchestrator.max_chunk_size", 100) v.SetDefault("orchestrator.cron_schedule", "0 0 2 5 * *") v.SetDefault("orchestrator.cron_timezone", "Asia/Jakarta") + v.SetDefault("orchestrator.stuck_chunk_timeout", 10*time.Minute) + v.SetDefault("orchestrator.stuck_sweep_interval", 2*time.Minute) v.SetDefault("tracing.enabled", false) v.SetDefault("tracing.service_name", "finance-cost-orchestrator") diff --git a/services/finance-cost-orchestrator/internal/orchestrator/coordinator.go b/services/finance-cost-orchestrator/internal/orchestrator/coordinator.go index 137b7a30..5b34f783 100644 --- a/services/finance-cost-orchestrator/internal/orchestrator/coordinator.go +++ b/services/finance-cost-orchestrator/internal/orchestrator/coordinator.go @@ -60,6 +60,7 @@ func (c *Coordinator) Run(ctx context.Context) error { chunkConsumer := rmq.NewConsumer(c.rmqConn, rmq.QueueChunkDone, "orchestrator-chunk-completed") go c.scrapeQueueDepth(ctx) + go c.sweepStuckChunks(ctx) errCh := make(chan error, 2) go func() { errCh <- jobConsumer.Consume(ctx, c.handleJobTriggered) }() @@ -367,6 +368,76 @@ func (c *Coordinator) scrapeQueueDepth(ctx context.Context) { } } +// sweepStuckChunks periodically finds DISPATCHED chunks that have been sitting +// longer than the configured timeout (default 10m) — these are chunks whose +// worker crashed (OOM, node eviction) without publishing ChunkCompletedEvent. +// For each stuck chunk it marks FAILED and publishes a synthetic completion +// event so advanceAfterChunk can progress the wave. +func (c *Coordinator) sweepStuckChunks(ctx context.Context) { + interval := c.cfg.Orchestrator.StuckSweepInterval + if interval <= 0 { + interval = 2 * time.Minute + } + timeout := c.cfg.Orchestrator.StuckChunkTimeout + if timeout <= 0 { + timeout = 10 * time.Minute + } + log.Info(). + Dur("interval", interval). + Dur("timeout", timeout). + Msg("stuck chunk sweeper started") + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + c.doSweep(ctx, timeout) + } + } +} + +func (c *Coordinator) doSweep(ctx context.Context, timeout time.Duration) { + stuck, err := c.chunkRepo.FindStuckChunks(ctx, timeout) + if err != nil { + log.Error().Err(err).Msg("stuck chunk sweep: query failed") + return + } + if len(stuck) == 0 { + return + } + log.Warn().Int("count", len(stuck)).Msg("stuck chunk sweep: found stuck chunks") + + for _, sc := range stuck { + if err := c.chunkRepo.MarkChunkFailed(ctx, sc.ChunkID, "stuck chunk timeout: worker likely crashed"); err != nil { + log.Error().Err(err).Int64("chunk_id", sc.ChunkID).Msg("stuck chunk sweep: mark failed") + continue + } + metrics.ChunksTotal.WithLabelValues(statusFailed).Inc() + + ev := ChunkCompletedEvent{ + ChunkID: sc.ChunkID, + JobID: sc.JobID, + WaveNo: sc.WaveNo, + Status: statusFailed, + SuccessCount: 0, + FailedCount: sc.ProductCount, + BlockedCount: 0, + } + if pubErr := c.pub.Publish(ctx, rmq.RoutingKeyChunkDone, ev); pubErr != nil { + log.Error().Err(pubErr).Int64("chunk_id", sc.ChunkID).Msg("stuck chunk sweep: publish completion failed") + continue + } + log.Info(). + Int64("chunk_id", sc.ChunkID). + Int64("job_id", sc.JobID). + Int("wave_no", sc.WaveNo). + Msg("stuck chunk sweep: marked failed + published completion") + } +} + // emitJobTerminal increments finance_cost_jobs_total. Best-effort; jobID is // only used to look up labels — if the lookup fails (e.g. row deleted) the // counter is incremented with empty labels rather than skipped, so totals diff --git a/services/finance-cost-orchestrator/internal/orchestrator/repos.go b/services/finance-cost-orchestrator/internal/orchestrator/repos.go index 0d13743b..bf3ca56e 100644 --- a/services/finance-cost-orchestrator/internal/orchestrator/repos.go +++ b/services/finance-cost-orchestrator/internal/orchestrator/repos.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "strings" + "time" "github.com/lib/pq" @@ -294,6 +295,61 @@ func (r *ChunkRepo) CountByJobWave(ctx context.Context, jobID int64, wave int) ( return } +// StuckChunkRow is a minimal struct returned by FindStuckChunks. +type StuckChunkRow struct { + ChunkID int64 + JobID int64 + WaveNo int + ProductCount int +} + +// FindStuckChunks returns DISPATCHED chunks whose dispatched_at is older than +// the given threshold. These chunks likely died (worker OOM / crash) without +// publishing a ChunkCompletedEvent. +func (r *ChunkRepo) FindStuckChunks(ctx context.Context, olderThan time.Duration) ([]*StuckChunkRow, error) { + const q = ` + SELECT cjc_chunk_id, cjc_job_id, cjc_wave_no, cjc_product_count + FROM cal_job_chunk + WHERE cjc_status = $1 + AND cjc_dispatched_at < now() - $2::interval + ORDER BY cjc_dispatched_at ASC + ` + rows, err := r.db.QueryContext(ctx, q, statusDispatched, fmt.Sprintf("%d seconds", int(olderThan.Seconds()))) + if err != nil { + return nil, fmt.Errorf("find stuck chunks: %w", err) + } + defer func() { + if e := rows.Close(); e != nil { + _ = e + } + }() + var out []*StuckChunkRow + for rows.Next() { + var c StuckChunkRow + if err := rows.Scan(&c.ChunkID, &c.JobID, &c.WaveNo, &c.ProductCount); err != nil { + return nil, fmt.Errorf("scan stuck chunk: %w", err) + } + out = append(out, &c) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate stuck chunks: %w", err) + } + return out, nil +} + +// MarkChunkFailed transitions a single chunk to FAILED with an error message. +func (r *ChunkRepo) MarkChunkFailed(ctx context.Context, chunkID int64, errMsg string) error { + const q = ` + UPDATE cal_job_chunk + SET cjc_status = $1, cjc_completed_at = now(), cjc_error_message = $2 + WHERE cjc_chunk_id = $3 AND cjc_status = $4 + ` + if _, err := r.db.ExecContext(ctx, q, statusFailed, errMsg, chunkID, statusDispatched); err != nil { + return fmt.Errorf("mark chunk %d failed: %w", chunkID, err) + } + return nil +} + // ListChunksOfWave returns the rows needed to re-publish ChunkSpec messages. func (r *ChunkRepo) ListChunksOfWave(ctx context.Context, jobID int64, wave int) ([]*ChunkRow, error) { rows, err := r.db.QueryContext(ctx, ` diff --git a/services/finance/internal/infrastructure/postgres/cost_result_repository.go b/services/finance/internal/infrastructure/postgres/cost_result_repository.go index 666649f9..9f17f10e 100644 --- a/services/finance/internal/infrastructure/postgres/cost_result_repository.go +++ b/services/finance/internal/infrastructure/postgres/cost_result_repository.go @@ -35,10 +35,19 @@ const resultColumns = `cpc_cost_id, cpc_product_sys_id, cpc_period, cpc_calculat COALESCE(cpc_job_id, 0), cpc_calculated_at, cpc_calculated_by, cpc_verified_at, COALESCE(cpc_verified_by, '')` +// upsertMaxRetries caps the retry loop when a concurrent transaction inserts a +// conflicting active row between our supersede and insert. +const upsertMaxRetries = 3 + // UpsertWithSupersede atomically SUPERSEDEs any existing active row for the // (product, period, calc_type) tuple, then inserts the new row with version // = prev+1. Returns the new cost id plus the previous (if any) version, total, // and id so the caller can write an audit-history row outside the transaction. +// +// When two concurrent jobs compute the same product+period+calc_type and no +// prior active row exists, the second INSERT hits uk_cpc_active. The retry +// loop rolls back, re-opens a fresh transaction (where the winner's row is now +// visible), supersedes it, and inserts again. func (r *CostResultRepository) UpsertWithSupersede( ctx context.Context, res *costcalc.Result, ) (newCostID int64, prevVersion int, prevTotal float64, prevCostID int64, err error) { @@ -49,6 +58,25 @@ func (r *CostResultRepository) UpsertWithSupersede( if res == nil { return 0, 0, 0, 0, fmt.Errorf("upsert result: nil result") } + + for attempt := range upsertMaxRetries { + newCostID, prevVersion, prevTotal, prevCostID, err = r.tryUpsert(ctx, res) + if err == nil { + return newCostID, prevVersion, prevTotal, prevCostID, nil + } + if !isUniqueViolation(err) { + return 0, 0, 0, 0, err + } + metrics.UpsertRetryTotal.Inc() + if attempt < upsertMaxRetries-1 { + continue + } + } + return 0, 0, 0, 0, fmt.Errorf("upsert result: unique violation after %d retries: %w", upsertMaxRetries, err) +} + +// tryUpsert runs one supersede+insert attempt inside a single transaction. +func (r *CostResultRepository) tryUpsert(ctx context.Context, res *costcalc.Result) (int64, int, float64, int64, error) { tx, err := r.db.BeginTx(ctx, nil) if err != nil { return 0, 0, 0, 0, fmt.Errorf("begin upsert tx: %w", err) @@ -63,12 +91,12 @@ func (r *CostResultRepository) UpsertWithSupersede( } }() - prevVersion, prevTotal, prevCostID, err = supersedePrevious(ctx, tx, res.ProductSysID(), res.Period(), res.CalcType()) + prevVersion, prevTotal, prevCostID, err := supersedePrevious(ctx, tx, res.ProductSysID(), res.Period(), res.CalcType()) if err != nil { return 0, 0, 0, 0, err } - newCostID, err = insertNewResult(ctx, tx, res, prevVersion+1) + newCostID, err := insertNewResult(ctx, tx, res, prevVersion+1) if err != nil { return 0, 0, 0, 0, err }