Revert "Improve shutdown handling from different replicator types (#106)" (#108)

This reverts commit 864b9d1.

Author: shayonj

.github/workflows/ci.yml

@@ -8,7 +8,6 @@ on:
 jobs:
   lint:
     runs-on: ubuntu-latest
-    timeout-minutes: 10
     steps:
       - uses: actions/checkout@v4
 
@@ -21,12 +20,10 @@ jobs:
         run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v2.4.0
 
       - name: Lint
-        timeout-minutes: 5
         run: make lint
 
   test:
     runs-on: ubuntu-latest
-    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v4
 
@@ -35,13 +32,11 @@ jobs:
         with:
           go-version: "1.24"
 
-      - name: Test with race detection
-        timeout-minutes: 10
+      - name: Test
         run: make test
   build:
     needs: [lint, test]
     runs-on: ubuntu-latest
-    timeout-minutes: 10
     steps:
       - uses: actions/checkout@v4
 
@@ -51,8 +46,7 @@ jobs:
           go-version: "1.24"
 
       - name: Build
-        timeout-minutes: 5
-        run: make build-ci
+        run: make build
 
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v3

.github/workflows/integration.yml

@@ -12,17 +12,15 @@ permissions:
 jobs:
   build:
     runs-on: ubuntu-latest
-    timeout-minutes: 10
     steps:
       - uses: actions/checkout@v4
 
       - name: Set up Go
         uses: actions/setup-go@v4
         with:
-          go-version: "1.24"
+          go-version: "1.21"
 
       - name: Build
-        timeout-minutes: 5
         run: make build
 
       - name: Upload binary
@@ -34,7 +32,6 @@ jobs:
   tests:
     needs: build
     runs-on: ubuntu-latest
-    timeout-minutes: 30
     strategy:
       fail-fast: false
       matrix:
@@ -72,7 +69,6 @@ jobs:
           sudo curl -L "https://github.com/docker/compose/releases/download/v2.17.2/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
           sudo chmod +x /usr/local/bin/docker-compose
       - name: Run test
-        timeout-minutes: 25
         env:
           PG_HOST: localhost
           PG_PORT: 5433
@@ -88,8 +84,8 @@ jobs:
           docker-compose -f internal/docker-compose.yml up -d
           sleep 10
           if [[ "${{ matrix.test }}" == "order" || "${{ matrix.test }}" == "resume" || "${{ matrix.test }}" == "postgres_uniqueness" ]]; then
-            timeout 20m ruby ./internal/scripts/e2e_${{ matrix.test }}_test.rb
+            ruby ./internal/scripts/e2e_${{ matrix.test }}_test.rb
           else
-            timeout 20m ./internal/scripts/e2e_${{ matrix.test }}.sh
+            ./internal/scripts/e2e_${{ matrix.test }}.sh
           fi
           docker-compose -f internal/docker-compose.yml down -v

Makefile

@@ -1,25 +1,23 @@
-.PHONY: test lint build build-ci build-prod clean check test-short
+.PHONY: test lint build clean
 
-.DEFAULT_GOAL := build-prod
+# Define the default goal
+.DEFAULT_GOAL := build
 
-build: build-prod
-
-build-ci:
-	go build -race -o bin/pg_flo
-
-build-prod:
+# Build the application
+build:
 	go build -o bin/pg_flo
 
+# Run tests with race detection and coverage
 test:
 	go test -v -race -coverprofile=coverage.txt -covermode=atomic ./...
 
-test-short:
-	go test -v -race -short -timeout=5m ./...
-
+# Run linter
 lint:
 	golangci-lint run --timeout=5m
 
+# Clean build artifacts
 clean:
 	rm -rf bin/ coverage.txt
 
+# Run all checks (lint and test)
 check: lint test

cmd/root.go

@@ -428,43 +428,18 @@ func runWorker(cmd *cobra.Command, _ []string) {
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
 
-	errCh := make(chan error, 1)
 	go func() {
-		errCh <- w.Start(ctx)
+		<-sigCh
+		cancel()
 	}()
 
 	log.Info().Msg("Starting worker...")
-
-	select {
-	case sig := <-sigCh:
-		log.Info().Str("signal", sig.String()).Msg("Received shutdown signal")
-
-		// Create a new context with timeout for graceful shutdown
-		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
-		defer shutdownCancel()
-
-		// Cancel the main context first to trigger graceful shutdown
-		cancel()
-
-		// Wait for worker to finish gracefully or timeout
-		select {
-		case err := <-errCh:
-			if err == context.Canceled {
-				log.Info().Msg("Worker shut down gracefully")
-			} else {
-				log.Error().Err(err).Msg("Worker encountered an error during shutdown")
-			}
-		case <-shutdownCtx.Done():
-			log.Error().Msg("Worker shutdown timeout exceeded")
-			os.Exit(1)
-		}
-
-	case err := <-errCh:
-		if err != nil {
-			log.Error().Err(err).Msg("Worker error occurred")
-			os.Exit(1)
+	if err := w.Start(ctx); err != nil {
+		if err == context.Canceled {
+			log.Info().Msg("Worker shut down gracefully")
+		} else {
+			log.Error().Err(err).Msg("Worker encountered an error during shutdown")
 		}
-		log.Info().Msg("Worker completed successfully")
 	}
 }
 

internal/scripts/e2e_postgres_data_type.sh

@@ -4,7 +4,7 @@ set -euo pipefail
 source "$(dirname "$0")/e2e_common.sh"
 
 create_test_table() {
-  log "Creating test table with various data types including   array testing..."
+  log "Creating test table with various data types including comprehensive array testing..."
 
   # Create required extensions
   run_sql "CREATE EXTENSION IF NOT EXISTS hstore;"

internal/scripts/e2e_test_local.sh

@@ -29,7 +29,7 @@ cleanup() {
 
 trap cleanup EXIT
 
-make build-ci
+make build
 
 setup_docker
 

pkg/pgflonats/pgflonats.go

@@ -49,21 +49,13 @@ func NewNATSClient(url, stream, group string) (*NATSClient, error) {
 		nats.MaxReconnects(-1),
 		nats.ReconnectWait(time.Second),
 		nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
-			if err != nil {
-				fmt.Printf("Disconnected due to: %v, will attempt reconnects\n", err)
-			} else {
-				fmt.Printf("Disconnected, will attempt reconnects\n")
-			}
+			fmt.Printf("Disconnected due to: %s, will attempt reconnects\n", err)
 		}),
 		nats.ReconnectHandler(func(nc *nats.Conn) {
 			fmt.Printf("Reconnected [%s]\n", nc.ConnectedUrl())
 		}),
 		nats.ClosedHandler(func(nc *nats.Conn) {
-			if lastErr := nc.LastError(); lastErr != nil {
-				fmt.Printf("Exiting: %v\n", lastErr)
-			} else {
-				fmt.Printf("Exiting: connection closed\n")
-			}
+			fmt.Printf("Exiting: %v\n", nc.LastError())
 		}),
 	)
 	if err != nil {
@@ -121,22 +113,9 @@ func (nc *NATSClient) PublishMessage(subject string, data []byte) error {
 	return nil
 }
 
-// Close gracefully closes the NATS connection by draining pending messages.
+// Close closes the NATS connection.
 func (nc *NATSClient) Close() error {
-	if nc.conn == nil {
-		return nil
-	}
-
-	if nc.conn.IsClosed() {
-		return nil
-	}
-
-	err := nc.conn.Drain()
-	if err != nil {
-		nc.conn.Close()
-		return fmt.Errorf("failed to drain NATS connection: %w", err)
-	}
-
+	nc.conn.Close()
 	return nil
 }