[+] add etcd<->postgres sync

Author: pashagolub

cmd/etcd_fdw/main.go

@@ -6,6 +6,8 @@ import (
 "context"
 "fmt"
 "os"
+"os/signal"
+"syscall"
 
 "github.com/jessevdk/go-flags"
 "github.com/sirupsen/logrus"
@@ -128,7 +130,19 @@ func main() {
 logrus.WithError(err).Fatal("Failed to setup logging")
 }
 
-ctx := context.Background()
+// Setup graceful shutdown
+ctx, cancel := context.WithCancel(context.Background())
+defer cancel()
+
+// Setup signal handling for graceful shutdown
+sigChan := make(chan os.Signal, 1)
+signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+
+go func() {
+sig := <-sigChan
+logrus.WithField("signal", sig).Info("Received shutdown signal, initiating graceful shutdown...")
+cancel()
+}()
 
 // Connect to PostgreSQL
 var pgPool db.PgxPoolIface
@@ -153,7 +167,9 @@ func main() {
 
 // Create and start sync service
 syncService := sync.NewService(pgPool, etcdClient, prefix, config.DryRun)
-if err := syncService.Start(ctx); err != nil {
+if err := syncService.Start(ctx); err != nil && ctx.Err() == nil {
 logrus.WithError(err).Fatal("Synchronization failed")
 }
+
+logrus.Info("Graceful shutdown completed")
 }

internal/db/postgres.go

@@ -220,11 +220,11 @@ func GetPendingWALEntries(ctx context.Context, pool PgxIface) ([]WALEntry, error
 return entries, nil
 }
 
-// DeleteWALEntry removes a processed WAL entry
-func DeleteWALEntry(ctx context.Context, pool PgxIface, key string, timestamp string) error {
-query := `DELETE FROM etcd_wal WHERE key = $1 AND ts = $2`
+// UpdateWALEntry removes a processed WAL entry
+func UpdateWALEntry(ctx context.Context, pool PgxIface, key string, timestamp string, revision int64) error {
+query := `UPDATE etcd_wal SET revision = $3 WHERE key = $1 AND ts = $2`
 
-_, err := pool.Exec(ctx, query, key, timestamp)
+_, err := pool.Exec(ctx, query, key, timestamp, revision)
 if err != nil {
 return fmt.Errorf("failed to delete WAL entry: %w", err)
 }

internal/db/postgres_test.go

@@ -72,7 +72,7 @@ func TestInsertWALEntry(t *testing.T) {
 
 // Set up mock expectations
 mock.ExpectExec("INSERT INTO etcd_wal").
-WithArgs(key, "value", 1).
+WithArgs(key, &value, &revision).
 WillReturnResult(pgxmock.NewResult("INSERT", 1))
 
 // Test the function
@@ -127,12 +127,12 @@ func TestDeleteWALEntry(t *testing.T) {
 ctx := context.Background()
 
 // Set up mock expectations
-mock.ExpectExec("DELETE FROM etcd_wal WHERE key").
-WithArgs("test/key", "2023-01-01T00:00:00Z").
-WillReturnResult(pgxmock.NewResult("DELETE", 1))
+mock.ExpectExec("UPDATE etcd_wal").
+WithArgs("test/key", "2023-01-01T00:00:00Z", int64(-1)).
+WillReturnResult(pgxmock.NewResult("UPDATE", 1))
 
 // Test the function
-err = DeleteWALEntry(ctx, mock, "test/key", "2023-01-01T00:00:00Z")
+err = UpdateWALEntry(ctx, mock, "test/key", "2023-01-01T00:00:00Z", -1)
 require.NoError(t, err)
 
 // Verify all expectations were met

internal/sync/resolver.go

@@ -3,16 +3,20 @@ package sync
 
 import (
 "context"
+"encoding/json"
 "fmt"
 "time"
 
+"github.com/jackc/pgx/v5/pgconn"
 "github.com/sirupsen/logrus"
 clientv3 "go.etcd.io/etcd/client/v3"
 
 "github.com/cybertec-postgresql/etcd_fdw/internal/db"
 "github.com/cybertec-postgresql/etcd_fdw/internal/etcd"
 )
 
+const InvalidRevision = -1
+
 // Service orchestrates bidirectional synchronization between etcd and PostgreSQL
 type Service struct {
 pgPool db.PgxPoolIface
@@ -224,25 +228,81 @@ func (s *Service) syncPostgreSQLToEtcd(ctx context.Context) error {
 }
 
 if err := s.processPostgreSQLNotification(ctx, notification); err != nil {
-logrus.WithError(err).WithField("payload", "unknown").Error("Failed to process PostgreSQL notification")
+logrus.WithError(err).WithField("payload", notification.Payload).Error("Failed to process PostgreSQL notification")
 // Continue processing other notifications rather than failing entirely
 }
 }
 }
 }
 
 // processPostgreSQLNotification processes a PostgreSQL NOTIFY and syncs to etcd
-func (s *Service) processPostgreSQLNotification(ctx context.Context, notification interface{}) error {
-// In a real implementation, we would parse the JSON payload to get WAL entry details
-// For now, we'll log that we received the notification
-logrus.WithField("notification", notification).Info("Received PostgreSQL notification")
-
-// TODO: Parse the notification and sync the change to etcd
-// This would involve:
-// 1. Parse notification JSON to get key, value, revision
-// 2. Apply conflict resolution logic
-// 3. Put/Delete to etcd
-// 4. Mark WAL entry as processed
+func (s *Service) processPostgreSQLNotification(ctx context.Context, notification *pgconn.Notification) error {
+// Parse the JSON payload from the notification
+var walNotification struct {
+Key string `json:"key"`
+Ts string `json:"ts"`
+Value *string `json:"value"`
+Revision *int64 `json:"revision"`
+Operation string `json:"operation"`
+}
 
-return nil
+if err := json.Unmarshal([]byte(notification.Payload), &walNotification); err != nil {
+return fmt.Errorf("failed to parse notification payload: %w", err)
+}
+
+logrus.WithFields(logrus.Fields{
+"key": walNotification.Key,
+"ts": walNotification.Ts,
+"operation": walNotification.Operation,
+}).Info("Processing PostgreSQL notification")
+
+// Apply conflict resolution: check if etcd has a newer version
+etcdKV, err := s.etcdClient.Get(ctx, walNotification.Key)
+if err != nil {
+return fmt.Errorf("failed to get key from etcd for conflict resolution: %w", err)
+}
+
+// Conflict resolution: etcd wins (if etcd has newer revision, skip this change)
+if etcdKV != nil && walNotification.Revision != nil && etcdKV.Revision > *walNotification.Revision {
+logrus.WithFields(logrus.Fields{
+"key": walNotification.Key,
+"etcd_revision": etcdKV.Revision,
+"local_revision": *walNotification.Revision,
+}).Warn("Conflict detected: etcd has newer revision, skipping local change")
+
+// Mark WAL entry as failed (conflict resolved - etcd wins)
+return db.UpdateWALEntry(ctx, s.pgPool, walNotification.Key, walNotification.Ts, InvalidRevision)
+}
+
+// Apply the change to etcd
+var newRevision int64
+switch walNotification.Operation {
+case "CREATE", "UPDATE":
+if walNotification.Value != nil {
+resp, err := s.etcdClient.Put(ctx, walNotification.Key, *walNotification.Value)
+if err != nil {
+// Mark WAL entry as failed
+db.UpdateWALEntry(ctx, s.pgPool, walNotification.Key, walNotification.Ts, InvalidRevision)
+return fmt.Errorf("failed to put key to etcd: %w", err)
+}
+newRevision = resp.Header.Revision
+logrus.WithField("key", walNotification.Key).Info("Synced PostgreSQL change to etcd (PUT)")
+}
+case "DELETE":
+resp, err := s.etcdClient.Delete(ctx, walNotification.Key)
+if err != nil {
+// Mark WAL entry as failed
+db.UpdateWALEntry(ctx, s.pgPool, walNotification.Key, walNotification.Ts, InvalidRevision)
+return fmt.Errorf("failed to delete key from etcd: %w", err)
+}
+newRevision = resp.Header.Revision
+logrus.WithField("key", walNotification.Key).Info("Synced PostgreSQL change to etcd (DELETE)")
+default:
+// Mark WAL entry as failed due to unknown operation
+db.UpdateWALEntry(ctx, s.pgPool, walNotification.Key, walNotification.Ts, InvalidRevision)
+return fmt.Errorf("unknown operation type: %s", walNotification.Operation)
+}
+
+// Mark WAL entry as successfully processed with the new etcd revision
+return db.UpdateWALEntry(ctx, s.pgPool, walNotification.Key, walNotification.Ts, newRevision)
 }