fix: Calc Engine Concurrent Upsert Race and Orchestrator OOM Deadlocks

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",

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")

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

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, `

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
 	}