Don't call ticker.Reset

flow/activities/flowable.go

@@ -721,8 +721,7 @@ func (a *FlowableActivity) SendWALHeartbeat(ctx context.Context) error {
 		return nil
 	}
 
-	sendTimeout := 10 * time.Minute
-	ticker := time.NewTicker(sendTimeout)
+	ticker := time.NewTicker(10 * time.Minute)
 	defer ticker.Stop()
 
 	activity.RecordHeartbeat(ctx, "sending walheartbeat every 10 minutes")
@@ -768,7 +767,6 @@ func (a *FlowableActivity) SendWALHeartbeat(ctx context.Context) error {
 				slog.InfoContext(ctx, fmt.Sprintf("sent walheartbeat to peer %v", pgPeer.Name))
 			}
 		}
-		ticker.Reset(sendTimeout)
 	}
 }
 

flow/activities/slot.go

@@ -75,8 +75,7 @@ func (a *FlowableActivity) recordSlotSizePeriodically(
 		return
 	}
 
-	timeout := 5 * time.Minute
-	ticker := time.NewTicker(timeout)
+	ticker := time.NewTicker(5 * time.Minute)
 	defer ticker.Stop()
 
 	for {
@@ -89,6 +88,5 @@ func (a *FlowableActivity) recordSlotSizePeriodically(
 		case <-ctx.Done():
 			return
 		}
-		ticker.Reset(timeout)
 	}
 }

flow/connectors/eventhub/eventhub.go

@@ -124,9 +124,7 @@ func (c *EventHubConnector) processBatch(
 	batchPerTopic := NewHubBatches(c.hubManager)
 	toJSONOpts := model.NewToJSONOptions(c.config.UnnestColumns, false)
 
-	eventHubFlushTimeout := peerdbenv.PeerDBEventhubFlushTimeoutSeconds()
-
-	ticker := time.NewTicker(eventHubFlushTimeout)
+	ticker := time.NewTicker(peerdbenv.PeerDBEventhubFlushTimeoutSeconds())
 	defer ticker.Stop()
 
 	lastSeenLSN := int64(0)
@@ -219,8 +217,6 @@ func (c *EventHubConnector) processBatch(
 					return 0, fmt.Errorf("failed to update last offset: %w", err)
 				}
 			}
-
-			ticker.Reset(eventHubFlushTimeout)
 		}
 	}
 }

flow/connectors/utils/heartbeat.go

@@ -29,7 +29,6 @@ func HeartbeatRoutine(
 			case <-ctx.Done():
 				return
 			case <-ticker.C:
-				ticker.Reset(15 * time.Second)
 			}
 		}
 	}()