Merge branch 'main' into pgwire-17

.github/workflows/ci.yml

@@ -12,7 +12,7 @@ jobs:
   build:
     strategy:
       matrix:
-        runner: [ubicloud-standard-8]
+        runner: [ubicloud-standard-8-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
     timeout-minutes: 30
     services:

.github/workflows/customer-docker.yml

@@ -12,7 +12,7 @@ jobs:
   docker-build:
     strategy:
       matrix:
-        runner: [ubicloud]
+        runner: [ubicloud-standard-2-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
     permissions:
       contents: read

.github/workflows/dev-docker.yml

@@ -10,7 +10,7 @@ jobs:
   docker-build:
     strategy:
       matrix:
-        runner: [ubicloud]
+        runner: [ubuntu-latest]
     runs-on: ${{ matrix.runner }}
     permissions:
       contents: read

.github/workflows/flow.yml

@@ -11,12 +11,12 @@ jobs:
   flow_test:
     strategy:
       matrix:
-        runner: [ubicloud-standard-8, ubuntu-latest]
+        runner: [ubicloud-standard-8-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
-    timeout-minutes: 30
+    timeout-minutes: 40
     services:
       pg_cdc:
-        image: postgis/postgis:15-3.4-alpine
+        image: imresamu/postgis:15-3.4-alpine
         ports:
           - 7132:5432
         env:

.github/workflows/golang-lint.yml

@@ -13,7 +13,7 @@ jobs:
       pull-requests: write
     strategy:
       matrix:
-        runner: [ubicloud]
+        runner: [ubicloud-standard-2-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout

.github/workflows/rust-lint.yml

@@ -13,7 +13,7 @@ jobs:
       pull-requests: write
     strategy:
       matrix:
-        runner: [ubicloud-standard-4]
+        runner: [ubicloud-standard-4-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout

.github/workflows/stable-docker.yml

@@ -9,7 +9,7 @@ jobs:
   docker-build:
     strategy:
       matrix:
-        runner: [ubicloud]
+        runner: [ubicloud-standard-2-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
     permissions:
       contents: read

.github/workflows/ui-build.yml

@@ -13,7 +13,7 @@ jobs:
     name: Build & Test UI
     strategy:
       matrix:
-        runner: [ubicloud]
+        runner: [ubicloud-standard-2-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout

.github/workflows/ui-lint.yml

@@ -17,7 +17,7 @@ jobs:
     name: Run UI linters
     strategy:
       matrix:
-        runner: [ubicloud]
+        runner: [ubicloud-standard-2-ubuntu-2204-arm]
     runs-on: ${{ matrix.runner }}
     steps:
       - name: checkout

.gitignore

@@ -7,3 +7,6 @@ tmp/
 private/
 nexus/server/tests/assets/*.json
 nexus/server/tests/results/actual/
+
+go.work
+go.work.sum

flow/activities/flowable.go

@@ -160,50 +160,55 @@ func (a *FlowableActivity) CreateNormalizedTable(
 	return conn.SetupNormalizedTables(config)
 }
 
+func (a *FlowableActivity) handleSlotInfo(
+	ctx context.Context,
+	srcConn connectors.CDCPullConnector,
+	slotName string,
+	peerName string,
+) error {
+	slotInfo, err := srcConn.GetSlotInfo(slotName)
+	if err != nil {
+		log.Warnf("warning: failed to get slot info: %v", err)
+		return err
+	}
+
+	if len(slotInfo) != 0 {
+		return a.CatalogMirrorMonitor.AppendSlotSizeInfo(ctx, peerName, slotInfo[0])
+	}
+	return nil
+}
+
 func (a *FlowableActivity) recordSlotSizePeriodically(
 	ctx context.Context,
 	srcConn connectors.CDCPullConnector,
 	slotName string,
-	done <-chan struct{},
 	peerName string,
 ) {
-
 	timeout := 10 * time.Minute
 	ticker := time.NewTicker(timeout)
 
 	defer ticker.Stop()
 	for {
-		slotInfo, err := srcConn.GetSlotInfo(slotName)
-		if err != nil {
-			log.Warnf("warning: failed to get slot info: %v", err)
-		}
-
-		if len(slotInfo) == 0 {
-			continue
-		}
-
 		select {
 		case <-ticker.C:
-			a.CatalogMirrorMonitor.AppendSlotSizeInfo(ctx, peerName, slotInfo[0])
-		case <-done:
-			a.CatalogMirrorMonitor.AppendSlotSizeInfo(ctx, peerName, slotInfo[0])
+			err := a.handleSlotInfo(ctx, srcConn, slotName, peerName)
+			if err != nil {
+				return
+			}
+		case <-ctx.Done():
+			return
 		}
 		ticker.Stop()
 		ticker = time.NewTicker(timeout)
 	}
-
 }
 
 // StartFlow implements StartFlow.
 func (a *FlowableActivity) StartFlow(ctx context.Context,
 	input *protos.StartFlowInput) (*model.SyncResponse, error) {
 	activity.RecordHeartbeat(ctx, "starting flow...")
-	done := make(chan struct{})
-	defer close(done)
 	conn := input.FlowConnectionConfigs
-
 	ctx = context.WithValue(ctx, shared.CDCMirrorMonitorKey, a.CatalogMirrorMonitor)
-
 	dstConn, err := connectors.GetCDCSyncConnector(ctx, conn.Destination)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get destination connector: %w", err)
@@ -246,7 +251,8 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 		slotNameForMetrics = input.FlowConnectionConfigs.ReplicationSlotName
 	}
 
-	go a.recordSlotSizePeriodically(ctx, srcConn, slotNameForMetrics, done, input.FlowConnectionConfigs.Source.Name)
+	go a.recordSlotSizePeriodically(errCtx, srcConn, slotNameForMetrics, input.FlowConnectionConfigs.Source.Name)
+
 	// start a goroutine to pull records from the source
 	errGroup.Go(func() error {
 		return srcConn.PullRecords(&model.PullRecordsRequest{
@@ -371,7 +377,6 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
 
 	pushedRecordsWithCount := fmt.Sprintf("pushed %d records", numRecords)
 	activity.RecordHeartbeat(ctx, pushedRecordsWithCount)
-	done <- struct{}{}
 
 	return res, nil
 }