better error logging around LoadSnapshotNameFromCatalog

PeerDB-io · Oct 22, 2024 · 986928a · 986928a
1 parent 0242dc2
commit 986928a
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 4 deletions.
diff --git a/flow/activities/flowable.go b/flow/activities/flowable.go
@@ -493,7 +493,8 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context,
 	if config.ParentMirrorName != "" {
 		_, snapshotName, _, err = shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, config.ParentMirrorName)
 		if err != nil {
-			return nil, err
+			a.Alerter.LogFlowError(ctx, "[GetQRepPartitions] "+config.FlowJobName, err)
+			return nil, fmt.Errorf("[GetQRepPartitions] failed to LoadSnapshotNameFromCatalog: %w", err)
 		}
 	}
 

diff --git a/flow/activities/flowable_core.go b/flow/activities/flowable_core.go
@@ -428,7 +428,8 @@ func replicateQRepPartition[TRead any, TWrite any, TSync connectors.QRepSyncConn
 		if config.ParentMirrorName != "" {
 			_, snapshotName, _, err = shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, config.ParentMirrorName)
 			if err != nil {
-				return err
+				a.Alerter.LogFlowError(ctx, "[replicateQRepPartition] "+config.FlowJobName, err)
+				return fmt.Errorf("[replicateQRepPartition] failed to LoadSnapshotNameFromCatalog: %w", err)
 			}
 		}
 
@@ -518,7 +519,8 @@ func replicateXminPartition[TRead any, TWrite any, TSync connectors.QRepSyncConn
 			var err error
 			_, snapshotName, _, err = shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, config.ParentMirrorName)
 			if err != nil {
-				return err
+				a.Alerter.LogFlowError(ctx, "[replicateXminPartition] "+config.FlowJobName, err)
+				return fmt.Errorf("[replicateXminPartition] failed to LoadSnapshotNameFromCatalog: %w", err)
 			}
 		}
 

diff --git a/flow/activities/snapshot_activity.go b/flow/activities/snapshot_activity.go
@@ -90,7 +90,7 @@ func (a *SnapshotActivity) SetupReplication(
 			"select pg_drop_replication_slot($1)",
 			slotName,
 		); err != nil && !shared.IsSQLStateError(err, pgerrcode.UndefinedObject) {
-			return fmt.Errorf("Failed to drop slot from previous run: %w", err)
+			return fmt.Errorf("failed to drop slot from previous run: %w", err)
 		}
 	}
 	if _, err := a.CatalogPool.Exec(ctx,
@@ -144,6 +144,9 @@ func (a *SnapshotActivity) LoadSupportsTidScan(
 	flowJobName string,
 ) (bool, error) {
 	_, _, supportsTidScan, err := shared.LoadSnapshotNameFromCatalog(ctx, a.CatalogPool, flowJobName)
+	if err != nil {
+		a.Alerter.LogFlowError(ctx, "[LoadSupportsTidScan] "+flowJobName, err)
+	}
 	return supportsTidScan, err
 }