Honor migrator flag and add useful debugging (#1640)

This is a straightforward bug fix that wires up an existing migration config flag that wasn't being passed through, plus adds debug logging

Author: mkysel

dev/local.env

@@ -46,7 +46,6 @@ export XMTPD_PAYER_PRIVATE_KEY=$XMTPD_SIGNER_PRIVATE_KEY
 
 # Migration Server Options
 # Not enabled by default. DB Reader connection string should be set accordingly.
-export XMTPD_MIGRATION_SERVER_ENABLE=false
 export XMTPD_MIGRATION_PAYER_PRIVATE_KEY=$XMTPD_SIGNER_PRIVATE_KEY
 export XMTPD_MIGRATION_NODE_SIGNING_KEY=$XMTPD_SIGNER_PRIVATE_KEY
 # export XMTPD_MIGRATION_DB_READER_CONNECTION_STRING="postgres://postgres:xmtp@localhost:8765/postgres?sslmode=disable"
@@ -57,7 +56,6 @@ export XMTPD_MIGRATION_NODE_SIGNING_KEY=$XMTPD_SIGNER_PRIVATE_KEY
 # export XMTPD_MIGRATION_DB_NAMESPACE="xmtp"
 
 # Migration Client Options
-export XMTPD_MIGRATION_CLIENT_ENABLE=false
 export XMTPD_MIGRATION_CLIENT_FROM_NODE_ID=100
 
 # Reflection Options - Enable gRPC reflection for grpcurl/buf curl

pkg/server/server.go

@@ -348,6 +348,10 @@ func NewBaseServer(
 			),
 			sync.WithPayerReportDomainSeparator(domainSeparator),
 			sync.WithClientMetrics(clientMetrics),
+			sync.WithMigration(sync.MigrationConfig{
+				Enable:     cfg.Options.MigrationClient.Enable,
+				FromNodeID: cfg.Options.MigrationClient.FromNodeID,
+			}),
 		)
 		if err != nil {
 			cfg.Logger.Error("failed to initialize sync server", zap.Error(err))

pkg/sync/sync_worker.go

@@ -79,6 +79,13 @@ func (s *syncWorker) start() error {
 		return err
 	}
 
+	if s.migration.Enable {
+		s.logger.Info(
+			"Migration client is enabled. Will migrate from migration originator",
+			utils.OriginatorIDField(s.migration.FromNodeID),
+		)
+	}
+
 	for _, node := range nodes {
 		s.subscribeToNode(node.NodeID)
 	}
@@ -362,6 +369,13 @@ func (s *syncWorker) setupStream(
 			migrator.WelcomeMessageOriginatorID,
 			migrator.KeyPackagesOriginatorID,
 		}
+		if s.logger.Core().Enabled(zap.DebugLevel) {
+			s.logger.Debug(
+				"requesting additional migrated payloads from originator node",
+				utils.OriginatorIDField(s.migration.FromNodeID),
+				zap.Any("originators", originatorNodeIDs),
+			)
+		}
 	}
 
 	stream, err := client.SubscribeEnvelopes(