Merge pull request #167 from ilramdhan/be/poy-rm-v2-chain-202607

fix(finance): retry RabbitMQ connect at startup and self-heal on drop

Author: ilramdhan

services/finance/cmd/server/main.go

@@ -120,7 +120,7 @@ func run() error { //nolint:gocognit,gocyclo // linear service wiring / DI setup
 	}
 
 	// Setup RabbitMQ (optional - graceful degradation for publisher)
-	rmqAdapter, costJobPub, closeRabbitMQ := setupRabbitMQ(cfg)
+	rmqAdapter, costJobPub, closeRabbitMQ := setupRabbitMQ(ctx, cfg)
 	defer closeRabbitMQ()
 
 	// Wrap into explicit interface values so that when RabbitMQ is unavailable
@@ -981,13 +981,27 @@ func startServers(ctx context.Context, cfg *config.Config,
 // graceful degradation). Returns the job-publisher adapter (oracle sync / RM
 // cost), the cost-calc job-trigger publisher (orchestrator hand-off), and a
 // close function for graceful shutdown.
-func setupRabbitMQ(cfg *config.Config) (*rabbitmq.JobPublisherAdapter, *rabbitmq.CostJobPublisher, func()) {
-	rmqConn, err := rabbitmq.NewConnection(cfg.RabbitMQ, log.Logger)
+func setupRabbitMQ(ctx context.Context, cfg *config.Config) (*rabbitmq.JobPublisherAdapter, *rabbitmq.CostJobPublisher, func()) {
+	const connectAttempts = 3
+
+	rmqConn, err := rabbitmq.NewConnectionWithRetry(cfg.RabbitMQ, log.Logger, connectAttempts)
 	if err != nil {
-		log.Warn().Err(err).Msg("Failed to connect to RabbitMQ, sync trigger will fail")
+		log.Warn().
+			Err(err).
+			Str("url", rabbitmq.SanitizeURL(cfg.RabbitMQ.URL)).
+			Int("attempts", connectAttempts).
+			Msg("Failed to connect to RabbitMQ after all attempts; the service will still start, " +
+				"but RM cost recalculate/export, cost sheet export, oracle sync trigger and " +
+				"multi-product calc scopes will fail until RabbitMQ is reachable and finance is restarted")
 		return nil, nil, func() {}
 	}
 
+	// Supervise redials on connection loss and swaps the live channel in place,
+	// so publishers recover mid-life instead of staying broken for the pod's
+	// whole lifetime. Bound to the root ctx: cancelled on shutdown.
+	superviseCtx, stopSupervisor := context.WithCancel(ctx)
+	go rmqConn.Supervise(superviseCtx)
+
 	publisher := rabbitmq.NewPublisher(rmqConn, log.Logger)
 	adapter := rabbitmq.NewJobPublisherAdapter(publisher, log.Logger)
 
@@ -998,6 +1012,7 @@ func setupRabbitMQ(cfg *config.Config) (*rabbitmq.JobPublisherAdapter, *rabbitmq
 	}
 
 	closeFunc := func() {
+		stopSupervisor()
 		if closeErr := rmqConn.Close(); closeErr != nil {
 			log.Warn().Err(closeErr).Msg("Failed to close RabbitMQ connection")
 		}

services/finance/cmd/worker/main.go

@@ -80,7 +80,7 @@ func run() error { //nolint:gocognit,gocyclo // linear setup function
 	}
 
 	// Setup RabbitMQ.
-	rmqConn, err := rabbitmq.NewConnection(cfg.RabbitMQ, log.Logger)
+	rmqConn, err := rabbitmq.NewConnectionWithRetry(cfg.RabbitMQ, log.Logger, 3)
 	if err != nil {
 		return err
 	}

services/finance/internal/application/costsheet/request_export_handler.go

@@ -14,6 +14,11 @@ import (
 	"github.com/mutugading/goapps-backend/services/finance/internal/domain/job"
 )
 
+// ErrPublisherUnavailable is returned when the finance service has no working
+// RabbitMQ publisher, so no export job can be queued.
+var ErrPublisherUnavailable = errors.New("message queue unavailable: RabbitMQ not connected " +
+	"(finance service could not reach the broker at startup; check RabbitMQ health and restart the finance service)")
+
 // maxExportProducts caps how many products a single export job renders.
 // Filter resolutions above this no longer truncate — they fan out into a
 // parent job plus N ≤maxExportProducts-sized child jobs (see Handle), each
@@ -212,7 +217,7 @@ func chunkIDs(ids []int64, size int) [][]int64 {
 // validate checks the fields Handle needs before doing any work.
 func (h *RequestExportHandler) validate(cmd RequestExportCommand) error {
 	if h.publisher == nil {
-		return fmt.Errorf("message queue unavailable: RabbitMQ not connected")
+		return ErrPublisherUnavailable
 	}
 	if cmd.Period == "" {
 		return fmt.Errorf("period is required")

services/finance/internal/application/oraclesync/trigger_handler.go

@@ -2,12 +2,18 @@ package oraclesync
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
 	"github.com/mutugading/goapps-backend/services/finance/internal/domain/job"
 )
 
+// ErrPublisherUnavailable is returned when the finance service has no working
+// RabbitMQ publisher, so no sync job can be queued.
+var ErrPublisherUnavailable = errors.New("message queue unavailable: RabbitMQ not connected " +
+	"(finance service could not reach the broker at startup; check RabbitMQ health and restart the finance service)")
+
 // TriggerCommand holds the input for triggering a sync job.
 type TriggerCommand struct {
 	Period    string
@@ -41,7 +47,7 @@ func NewTriggerHandler(jobRepo job.Repository, publisher JobPublisher) *TriggerH
 // Handle creates a job execution and publishes it to the queue.
 func (h *TriggerHandler) Handle(ctx context.Context, cmd TriggerCommand) (*TriggerResult, error) {
 	if h.publisher == nil {
-		return nil, fmt.Errorf("message queue unavailable: RabbitMQ not connected")
+		return nil, ErrPublisherUnavailable
 	}
 
 	// Resolve period if not provided.

services/finance/internal/application/rmcost/request_export_handler.go

@@ -43,7 +43,7 @@ func NewRequestExportHandler(jobRepo job.Repository, publisher ExportJobPublishe
 // Handle creates a job_execution row and publishes the message to RabbitMQ.
 func (h *RequestExportHandler) Handle(ctx context.Context, cmd RequestExportCommand) (*RequestExportResult, error) {
 	if h.publisher == nil {
-		return nil, fmt.Errorf("message queue unavailable: RabbitMQ not connected")
+		return nil, ErrPublisherUnavailable
 	}
 	if cmd.Period == "" {
 		return nil, fmt.Errorf("period is required")

services/finance/internal/application/rmcost/trigger_handler.go

@@ -3,6 +3,7 @@ package rmcost
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -13,6 +14,12 @@ import (
 	"github.com/mutugading/goapps-backend/services/finance/internal/domain/rmcost"
 )
 
+// ErrPublisherUnavailable is returned when the finance service has no working
+// RabbitMQ publisher, so no job can be queued. The "RabbitMQ not connected"
+// substring is asserted on by handlers_test.go — keep it.
+var ErrPublisherUnavailable = errors.New("message queue unavailable: RabbitMQ not connected " +
+	"(finance service could not reach the broker at startup; check RabbitMQ health and restart the finance service)")
+
 // TriggerReason identifies why the calculation was requested. Maps to
 // rmcost.HistoryTriggerReason on the worker side via the job params JSON.
 type TriggerReason string
@@ -70,7 +77,7 @@ func NewTriggerHandler(jobRepo job.Repository, publisher JobPublisher) *TriggerH
 // so operators see the error instead of a permanently-QUEUED row.
 func (h *TriggerHandler) Handle(ctx context.Context, cmd TriggerCommand) (*TriggerResult, error) {
 	if h.publisher == nil {
-		return nil, fmt.Errorf("message queue unavailable: RabbitMQ not connected")
+		return nil, ErrPublisherUnavailable
 	}
 	if cmd.CreatedBy == "" {
 		return nil, rmcost.ErrEmptyCreatedBy