check for peer table existence

Author: Amogh-Bharadwaj

.github/workflows/flow.yml

@@ -92,7 +92,7 @@ jobs:
         env:
           PG_CDC: empty
           PGPASSWORD: postgres
-          
+
       - name: run tests
         run: |
           gotestsum --format testname -- -p 8 ./... -timeout 2400s

flow/activities/flowable.go

@@ -706,24 +706,24 @@ func (a *FlowableActivity) SendWALHeartbeat(ctx context.Context) error {
 	sendTimeout := 10 * time.Minute
 	ticker := time.NewTicker(sendTimeout)
 	defer ticker.Stop()
-
-	peerTableExists := true
-	pgPeers, err := getPostgresPeerConfigs(ctx)
-	if err != nil {
-		if strings.Contains(err.Error(), "does not exist") {
-			log.Warn("[sendwalheartbeat]: warning: no postgres peers found in catalog. Now I will be sleeping repeatedly")
-			peerTableExists = false
-		}
-	}
-
 	activity.RecordHeartbeat(ctx, "sending walheartbeat every 10 minutes")
 	for {
 		select {
 		case <-ctx.Done():
 			log.Info("context is done, exiting wal heartbeat send loop")
 			return nil
 		case <-ticker.C:
-			if peerTableExists {
+			peersTableExists := true
+			pgPeers, err := getPostgresPeerConfigs(ctx)
+			if err != nil {
+				if strings.Contains(err.Error(), "does not exist") {
+					log.Warn("[sendwalheartbeat]: warning: peers table not found. skipping walheartbeat send.")
+					peersTableExists = false
+				}
+				return fmt.Errorf("error getting postgres peers: %w", err)
+			}
+
+			if peersTableExists {
 				command := `
 				BEGIN;
 				DROP aggregate IF EXISTS PEERDB_EPHEMERAL_HEARTBEAT(float4);
@@ -753,9 +753,9 @@ func (a *FlowableActivity) SendWALHeartbeat(ctx context.Context) error {
 					log.Infof("sent walheartbeat to peer %v", pgPeer.Name)
 				}
 			}
-			ticker.Stop()
-			ticker = time.NewTicker(sendTimeout)
 		}
+		ticker.Stop()
+		ticker = time.NewTicker(sendTimeout)
 	}
 }