utils.RecordHeartbeat: use activity.IsActivity (#1186)

Ideally e2e tests would pass context with mock activity,
but in meantime this'll reduce noise in test logs
& is a bit more efficient

flow/connectors/bigquery/bigquery.go

@@ -908,7 +908,7 @@ func (c *BigQueryConnector) SetupNormalizedTables(
 		datasetTablesSet[*datasetTable] = struct{}{}
 		// log that table was created
 		c.logger.Info(fmt.Sprintf("created table %s", tableIdentifier))
-		utils.RecordHeartbeatWithRecover(c.ctx, fmt.Sprintf("created table %s", tableIdentifier))
+		utils.RecordHeartbeat(c.ctx, fmt.Sprintf("created table %s", tableIdentifier))
 	}
 
 	return &protos.SetupNormalizedTableBatchOutput{

flow/connectors/postgres/cdc.go

@@ -329,7 +329,7 @@ func (p *PostgresCDCSource) consumeStream(
 		rawMsg, err := conn.ReceiveMessage(ctx)
 		cancel()
 
-		utils.RecordHeartbeatWithRecover(p.ctx, "consumeStream ReceiveMessage")
+		utils.RecordHeartbeat(p.ctx, "consumeStream ReceiveMessage")
 		ctxErr := p.ctx.Err()
 		if ctxErr != nil {
 			return fmt.Errorf("consumeStream preempted: %w", ctxErr)

flow/connectors/postgres/postgres.go

@@ -569,7 +569,7 @@ func (c *PostgresConnector) GetTableSchema(
 			return nil, err
 		}
 		res[tableName] = tableSchema
-		utils.RecordHeartbeatWithRecover(c.ctx, fmt.Sprintf("fetched schema for table %s", tableName))
+		utils.RecordHeartbeat(c.ctx, fmt.Sprintf("fetched schema for table %s", tableName))
 		c.logger.Info(fmt.Sprintf("fetched schema for table %s", tableName))
 	}
 
@@ -681,7 +681,7 @@ func (c *PostgresConnector) SetupNormalizedTables(req *protos.SetupNormalizedTab
 
 		tableExistsMapping[tableIdentifier] = false
 		c.logger.Info(fmt.Sprintf("created table %s", tableIdentifier))
-		utils.RecordHeartbeatWithRecover(c.ctx, fmt.Sprintf("created table %s", tableIdentifier))
+		utils.RecordHeartbeat(c.ctx, fmt.Sprintf("created table %s", tableIdentifier))
 	}
 
 	err = createNormalizedTablesTx.Commit(c.ctx)
@@ -771,7 +771,7 @@ func (c *PostgresConnector) EnsurePullability(
 
 		if !req.CheckConstraints {
 			msg := fmt.Sprintf("[no-constraints] ensured pullability table %s", tableName)
-			utils.RecordHeartbeatWithRecover(c.ctx, msg)
+			utils.RecordHeartbeat(c.ctx, msg)
 			continue
 		}
 
@@ -791,7 +791,7 @@ func (c *PostgresConnector) EnsurePullability(
 			return nil, fmt.Errorf("table %s has no primary keys and does not have REPLICA IDENTITY FULL", schemaTable)
 		}
 
-		utils.RecordHeartbeatWithRecover(c.ctx, fmt.Sprintf("ensured pullability table %s", tableName))
+		utils.RecordHeartbeat(c.ctx, fmt.Sprintf("ensured pullability table %s", tableName))
 	}
 
 	return &protos.EnsurePullabilityBatchOutput{TableIdentifierMapping: tableIdentifierMapping}, nil

flow/connectors/snowflake/snowflake.go

@@ -293,7 +293,7 @@ func (c *SnowflakeConnector) GetTableSchema(
 			return nil, err
 		}
 		res[tableName] = tableSchema
-		utils.RecordHeartbeatWithRecover(c.ctx, fmt.Sprintf("fetched schema for table %s", tableName))
+		utils.RecordHeartbeat(c.ctx, fmt.Sprintf("fetched schema for table %s", tableName))
 	}
 
 	return &protos.GetTableSchemaBatchOutput{
@@ -461,7 +461,7 @@ func (c *SnowflakeConnector) SetupNormalizedTables(
 			return nil, fmt.Errorf("[sf] error while creating normalized table: %w", err)
 		}
 		tableExistsMapping[tableIdentifier] = false
-		utils.RecordHeartbeatWithRecover(c.ctx, fmt.Sprintf("created table %s", tableIdentifier))
+		utils.RecordHeartbeat(c.ctx, fmt.Sprintf("created table %s", tableIdentifier))
 	}
 
 	return &protos.SetupNormalizedTableBatchOutput{

flow/connectors/utils/heartbeat.go

@@ -3,7 +3,6 @@ package utils
 import (
 	"context"
 	"fmt"
-	"log/slog"
 	"time"
 
 	"go.temporal.io/sdk/activity"
@@ -22,7 +21,7 @@ func HeartbeatRoutine(
 		for {
 			counter += 1
 			msg := fmt.Sprintf("heartbeat #%d: %s", counter, message())
-			RecordHeartbeatWithRecover(ctx, msg)
+			RecordHeartbeat(ctx, msg)
 			select {
 			case <-shutdown:
 				return
@@ -37,12 +36,8 @@ func HeartbeatRoutine(
 
 // if the functions are being called outside the context of a Temporal workflow,
 // activity.RecordHeartbeat panics, this is a bandaid for that.
-func RecordHeartbeatWithRecover(ctx context.Context, details ...interface{}) {
-	defer func() {
-		if r := recover(); r != nil {
-			slog.Warn("ignoring panic from activity.RecordHeartbeat")
-			slog.Warn("this can happen when function is invoked outside of a Temporal workflow")
-		}
-	}()
-	activity.RecordHeartbeat(ctx, details...)
+func RecordHeartbeat(ctx context.Context, details ...interface{}) {
+	if activity.IsActivity(ctx) {
+		activity.RecordHeartbeat(ctx, details...)
+	}
 }