Revert Sentry suppressions; re-resolve pantheon hash at store (SERVICES-2C)

apps/hermes/worker.go

@@ -10,9 +10,6 @@ import (
 	"raidhub/lib/monitoring/hermes_metrics"
 	"raidhub/lib/utils"
 	"raidhub/lib/utils/logging"
-	"raidhub/lib/utils/network"
-	"raidhub/lib/utils/retry"
-	"raidhub/lib/web/discord"
 
 	amqp "github.com/rabbitmq/amqp091-go"
 )
@@ -61,20 +58,17 @@ func (w *Worker) Run() {
 			return
 		case msg, ok := <-w.channel:
 			if !ok {
-				cause := context.Cause(w.ctx)
-				if cause != nil && cause.Error() == AUTOSCALE_IN {
-					w.Debug(WORKER_STOPPING, map[string]any{
-						logging.REASON: AUTOSCALE_IN,
-					})
-				} else if w.ctx.Err() != nil {
+				// Check if this is a natural shutdown (context cancelled) or unexpected channel closure
+				select {
+				case <-w.ctx.Done():
+					// Natural shutdown - context was cancelled (e.g., autoscale, app shutdown)
 					w.Debug(WORKER_STOPPING, map[string]any{
 						logging.REASON: "channel_closed",
 					})
-				} else {
-					// Delivery channel can close during scale-in before cancel is selected.
-					w.Warn(WORKER_STOPPING, fmt.Errorf("channel_closed"), map[string]any{
-						logging.REASON: "delivery_channel_closed",
-					})
+				default:
+					// Unexpected channel closure - report as error
+					err := fmt.Errorf("channel_closed")
+					w.Error(WORKER_STOPPING, err, nil)
 				}
 				return
 			}
@@ -245,7 +239,7 @@ func (w *Worker) dropMessage(msg amqp.Delivery, retryCount int, maxRetries int,
 	if msg.Exchange != "" {
 		fields["exchange"] = msg.Exchange
 	}
-	w.logMessageFailure("MESSAGE_EXCEEDED_MAX_RETRIES", processingErr, fields)
+	w.Error("MESSAGE_EXCEEDED_MAX_RETRIES", processingErr, fields)
 
 	// Nack with requeue=false to permanently drop the message
 	// This prevents infinite retry loops
@@ -318,35 +312,5 @@ func (w *Worker) logUnretryableMessage(msg amqp.Delivery, err error) {
 	if originalErr := errors.Unwrap(err); originalErr != nil {
 		fields["original_error"] = originalErr.Error()
 	}
-	w.logMessageFailure("MESSAGE_UNRETRYABLE", err, fields)
-}
-
-func (w *Worker) logMessageFailure(key string, err error, fields map[string]any) {
-	if isOperationalMessageFailure(err) {
-		w.Warn(key, err, fields)
-		return
-	}
-	w.Error(key, err, fields)
-}
-
-func isOperationalMessageFailure(err error) bool {
-	if processing.IsUnretryableError(err) {
-		return true
-	}
-	if discord.IsPermanentDeliveryError(err) {
-		return true
-	}
-	var maxRetriesErr *retry.MaxRetriesExceededError
-	if errors.As(err, &maxRetriesErr) {
-		if network.IsCloudflareError(maxRetriesErr) ||
-			network.IsTimeout(maxRetriesErr) ||
-			network.IsConnectionError(maxRetriesErr) {
-			return true
-		}
-		if netErr := network.CategorizeNetworkError(maxRetriesErr); netErr != nil &&
-			netErr.Type == network.ErrorTypeServerError {
-			return true
-		}
-	}
-	return false
+	w.Error("MESSAGE_UNRETRYABLE", err, fields)
 }

lib/messaging/queue-workers/activity_history.go

@@ -45,8 +45,7 @@ func processActivityHistory(worker processing.WorkerInterface, message amqp.Deli
 	err = player.UpdateActivityHistory(worker.Context(), membershipId)
 
 	if err != nil {
-		// Worker logs MESSAGE_PROCESSING_ERROR at Warn and handles retries; avoid duplicate Sentry Error.
-		worker.Warn("ACTIVITY_HISTORY_PROCESSING_ERROR", err, map[string]any{
+		worker.Error("ACTIVITY_HISTORY_PROCESSING_ERROR", err, map[string]any{
 			logging.MEMBERSHIP_ID: membershipId,
 		})
 		return err

lib/services/instance_storage/orchestrator.go

@@ -7,6 +7,7 @@ import (
 	"raidhub/lib/messaging/publishing"
 	"raidhub/lib/messaging/routing"
 	"raidhub/lib/monitoring/global_metrics"
+	"raidhub/lib/services/pgcr_processing"
 	"raidhub/lib/services/subscriptions"
 	"raidhub/lib/utils/logging"
 	"raidhub/lib/web/bungie"
@@ -23,6 +24,10 @@ var logger = logging.NewLogger("INSTANCE_STORAGE_SERVICE")
 func StorePGCR(ctx context.Context, inst *dto.Instance, raw *bungie.DestinyPostGameCarnageReport) (*time.Duration, bool, error) {
 	startTime := time.Now()
 
+	if raw != nil {
+		inst.Hash = pgcr_processing.ResolveInstanceActivityHash(raw.ActivityDetails)
+	}
+
 	// Start transaction for atomic storage of pgcr + instance data
 	tx, err := postgres.DB.Begin()
 	if err != nil {

lib/services/pgcr_processing/process-pgcr.go

@@ -108,7 +108,7 @@ func parsePGCRToInstance(report *bungie.DestinyPostGameCarnageReport) (*dto.Inst
 
 	completionReason := getStat(report.Entries[0].Values, "completionReason")
 
-	activityHash := resolveInstanceActivityHash(report.ActivityDetails)
+	activityHash := ResolveInstanceActivityHash(report.ActivityDetails)
 
 	result := dto.Instance{
 		InstanceId: report.ActivityDetails.InstanceId,
@@ -327,10 +327,10 @@ var leviHashes = map[uint32]bool{
 	3879860661: true, 3857338478: true,
 }
 
-// resolveInstanceActivityHash returns the activity hash to store on the instance row.
+// ResolveInstanceActivityHash returns the activity hash to store on the instance row.
 // Pantheon featured-reprise playlists report the playlist wrapper in directorActivityHash
 // and the actual encounter in referenceId. For typical raids both fields match.
-func resolveInstanceActivityHash(ad bungie.DestinyHistoricalStatsActivity) uint32 {
+func ResolveInstanceActivityHash(ad bungie.DestinyHistoricalStatsActivity) uint32 {
 	if ad.ReferenceId != 0 && ad.ReferenceId != ad.DirectorActivityHash {
 		return ad.ReferenceId
 	}
@@ -367,7 +367,7 @@ func isFresh(pgcr *bungie.DestinyPostGameCarnageReport, deathless bool) (*bool,
 		// Pre beyond light, using StartingPhaseIndex
 		result = new(bool)
 		startingPhaseIndex := *pgcr.StartingPhaseIndex
-		activityHash := resolveInstanceActivityHash(pgcr.ActivityDetails)
+		activityHash := ResolveInstanceActivityHash(pgcr.ActivityDetails)
 		// sotp
 		if activityHash == 548750096 || activityHash == 2812525063 {
 			*result = (startingPhaseIndex <= 1)