log how long snapshot export maintained, & when snapshot flow fails (#…

…1352)

Mix up had #1347 merge before logging feedback implemented

flow/activities/snapshot_activity.go

@@ -124,8 +124,12 @@ func (a *SnapshotActivity) MaintainTx(ctx context.Context, sessionID string, pee
 	a.SnapshotConnections[sessionID] = sss
 	a.SnapshotConnectionsMutex.Unlock()
 
+	logger := activity.GetLogger(ctx)
+	start := time.Now()
 	for {
-		activity.RecordHeartbeat(ctx, "maintaining export snapshot transaction")
+		msg := fmt.Sprintf("maintaining export snapshot transaction %s", time.Since(start).Round(time.Second))
+		logger.Info(msg)
+		activity.RecordHeartbeat(ctx, msg)
 		if ctx.Err() != nil {
 			a.SnapshotConnectionsMutex.Lock()
 			delete(a.SnapshotConnections, sessionID)

flow/workflows/cdc_flow.go

@@ -307,6 +307,7 @@ func CDCFlowWorkflowWithConfig(
 		snapshotFlowCtx := workflow.WithChildOptions(ctx, childSnapshotFlowOpts)
 		snapshotFlowFuture := workflow.ExecuteChildWorkflow(snapshotFlowCtx, SnapshotFlowWorkflow, cfg)
 		if err := snapshotFlowFuture.Get(snapshotFlowCtx, nil); err != nil {
+			w.logger.Error("snapshot flow failed", slog.Any("error", err))
 			return state, fmt.Errorf("failed to execute child workflow: %w", err)
 		}