diff --git a/server/logrepl/replication.go b/server/logrepl/replication.go
index e8f99dc731..2d2dd0eb64 100755
--- a/server/logrepl/replication.go
+++ b/server/logrepl/replication.go
@@ -16,8 +16,10 @@ package logrepl
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
+	"os"
 	"strings"
 	"sync"
 	"time"
@@ -41,8 +43,10 @@ type rcvMsg struct {
 type LogicalReplicator struct {
 	primaryDns      string
 	replicationConn *pgx.Conn
-	receiveMsgChan  chan rcvMsg
+
+	walFilePath     string
 	running         bool
+	messageReceived bool
 	stop            chan struct{}
 	mu              *sync.Mutex
 }
@@ -50,7 +54,7 @@ type LogicalReplicator struct {
 // NewLogicalReplicator creates a new logical replicator instance which connects to the primary and replication
 // databases using the connection strings provided. The connection to the replica is established immediately, and the
 // connection to the primary is established when StartReplication is called.
-func NewLogicalReplicator(primaryDns string, replicationDns string) (*LogicalReplicator, error) {
+func NewLogicalReplicator(walFilePath string, primaryDns string, replicationDns string) (*LogicalReplicator, error) {
 	conn, err := pgx.Connect(context.Background(), replicationDns)
 	if err != nil {
 		return nil, err
@@ -59,35 +63,84 @@ func NewLogicalReplicator(primaryDns string, replicationDns string) (*LogicalRep
 	return &LogicalReplicator{
 		primaryDns:      primaryDns,
 		replicationConn: conn,
-		stop:            make(chan struct{}),
-		receiveMsgChan:  make(chan rcvMsg),
+		walFilePath:     walFilePath,
 		mu:              &sync.Mutex{},
 	}, nil
 }
 
-// SetupReplication sets up the replication slot and publication for the given database.
-func SetupReplication(primaryConnectionString string, publicationName string) error {
-	conn, err := pgconn.Connect(context.Background(), primaryConnectionString)
+// PrimaryDns returns the DNS for the primary database. Not suitable for RPCs used in replication e.g.
+// StartReplication. See ReplicationDns.
+func (r *LogicalReplicator) PrimaryDns() string {
+	return r.primaryDns
+}
+
+// ReplicationDns returns the DNS for the primary database with the replication query parameter appended. Not suitable
+// for normal query RPCs.
+func (r *LogicalReplicator) ReplicationDns() string {
+	if strings.Contains(r.primaryDns, "?") {
+		return fmt.Sprintf("%s&replication=database", r.primaryDns)
+	}
+	return fmt.Sprintf("%s?replication=database", r.primaryDns)
+}
+
+// CaughtUp returns true if the replication slot is caught up to the primary, and false otherwise. This only works if
+// there is only a single replication slot on the primary, so it's only suitable for testing.
+func (r *LogicalReplicator) CaughtUp() (bool, error) {
+	r.mu.Lock()
+	if !r.messageReceived {
+		r.mu.Unlock()
+		// We can't query the replication state until after receiving our first message
+		return false, nil
+	}
+	r.mu.Unlock()
+
+	conn, err := pgx.Connect(context.Background(), r.PrimaryDns())
 	if err != nil {
-		return err
+		return false, err
 	}
 	defer conn.Close(context.Background())
 
-	result := conn.Exec(context.Background(), fmt.Sprintf("DROP PUBLICATION IF EXISTS %s;", publicationName))
-	_, err = result.ReadAll()
+	result, err := conn.Query(context.Background(), "SELECT pg_wal_lsn_diff(write_lsn, sent_lsn) AS replication_lag FROM pg_stat_replication")
 	if err != nil {
-		return err
+		return false, err
 	}
 
-	result = conn.Exec(context.Background(), fmt.Sprintf("CREATE PUBLICATION %s FOR ALL TABLES;", publicationName))
-	_, err = result.ReadAll()
-	return err
+	defer result.Close()
+
+	for result.Next() {
+		rows, err := result.Values()
+		if err != nil {
+			return false, err
+		}
+
+		row := rows[0]
+		lag, ok := row.(pgtype.Numeric)
+		if ok && lag.Valid {
+			log.Printf("Current replication lag: %v", row)
+			return lag.Int.Int64() >= 0, nil
+		} else {
+			log.Printf("Replication lag unknown: %v", row)
+		}
+	}
+
+	if result.Err() != nil {
+		return false, result.Err()
+	}
+
+	// if we got this far, then there is no running replication thread, which we interpret as caught up
+	return true, nil
 }
 
+// maxConsecutiveFailures is the maximum number of consecutive RPC errors that can occur before we stop
+// the replication thread
+const maxConsecutiveFailures = 10
+
+var errShutdownRequested = errors.New("shutdown requested")
+
 // StartReplication starts the replication process for the given slot name. This function blocks until replication is
 // stopped via the Stop method, or an error occurs.
 func (r *LogicalReplicator) StartReplication(slotName string) error {
-	standbyMessageTimeout := time.Second * 10
+	standbyMessageTimeout := 10 * time.Second
 	nextStandbyMessageDeadline := time.Now().Add(standbyMessageTimeout)
 	relationsV2 := map[uint32]*pglogrepl.RelationMessageV2{}
 	typeMap := pgtype.NewMap()
@@ -96,149 +149,217 @@ func (r *LogicalReplicator) StartReplication(slotName string) error {
 	// on StreamStopMessage we set it back to false
 	inStream := false
 
-	// We fail after 3 consecutive network errors excluding timeouts. Any successful RPC resets the counter.
-	connErrCnt := 0
-	var primaryConn *pgconn.PgConn
-	var clientXLogPos pglogrepl.LSN
+	// lsn is the last WAL position we have received from the server, which we send back to the server via
+	// SendStandbyStatusUpdate after every message we get. Postgres tracks this LSN for each slot, which allows us to
+	// resume where we left off in the case of an interruption.
+	var lsn pglogrepl.LSN
+	lsn, err := r.readWALPosition()
+	if err != nil {
+		return err
+	}
 
+	var primaryConn *pgconn.PgConn
 	defer func() {
 		if primaryConn != nil {
 			_ = primaryConn.Close(context.Background())
 		}
-		r.mu.Lock()
-		r.running = false
-		r.mu.Unlock()
+		// We always shut down here and only here, so we do the cleanup on thread exit in exactly one place
+		r.shutdown()
 	}()
 
+	connErrCnt := 0
+	handleErrWithRetry := func(err error) error {
+		if err != nil {
+			connErrCnt++
+			if connErrCnt < maxConsecutiveFailures {
+				log.Printf("Error: %v. Retrying", err)
+				_ = primaryConn.Close(context.Background())
+				primaryConn = nil
+				return nil
+			}
+		} else {
+			connErrCnt = 0
+		}
+
+		return err
+	}
+
+	sendStandbyStatusUpdate := func(currentLSN pglogrepl.LSN) error {
+		// The StatusUpdate message wants us to respond with the current position in the WAL + 1:
+		// https://www.postgresql.org/docs/current/protocol-replication.html
+		lsn := currentLSN + 1
+		err := pglogrepl.SendStandbyStatusUpdate(context.Background(), primaryConn, pglogrepl.StandbyStatusUpdate{WALWritePosition: lsn + 1})
+		if err != nil {
+			return handleErrWithRetry(err)
+		}
+
+		log.Printf("Sent Standby status message at %s\n", lsn.String())
+		nextStandbyMessageDeadline = time.Now().Add(standbyMessageTimeout)
+		return nil
+	}
+
+	log.Println("Starting replicator")
 	r.mu.Lock()
 	r.running = true
+	r.stop = make(chan struct{})
 	r.mu.Unlock()
 
 	for {
+		err := func() error {
+			// Shutdown if requested
+			select {
+			case <-r.stop:
+				return errShutdownRequested
+			default:
+				// continue below
+			}
 
-		// Shutdown if requested
-		select {
-		case <-r.stop:
-			r.shutdown()
-			return nil
-		default:
-			// continue
-		}
-
-		if primaryConn == nil {
-			// TODO: not sure if this retry logic is correct, with some failures we appear to miss events that aren't
-			//  sent again
-			var err error
-			primaryConn, clientXLogPos, err = r.beginReplication(slotName)
-			if err != nil {
-				return err
+			if primaryConn == nil {
+				var err error
+				primaryConn, err = r.beginReplication(slotName, lsn)
+				if err != nil {
+					// unlike other error cases, back off a little here, since we're likely to just get the same error again
+					// on initial replication establishment
+					time.Sleep(100 * time.Millisecond)
+					return handleErrWithRetry(err)
+				}
 			}
-		}
 
-		if time.Now().After(nextStandbyMessageDeadline) {
-			err := pglogrepl.SendStandbyStatusUpdate(context.Background(), primaryConn, pglogrepl.StandbyStatusUpdate{WALWritePosition: clientXLogPos})
-			if err != nil {
-				connErrCnt++
-				if connErrCnt < 3 {
-					// re-establish connection on next pass through the loop
-					_ = primaryConn.Close(context.Background())
-					primaryConn = nil
-					continue
+			if time.Now().After(nextStandbyMessageDeadline) {
+				err := sendStandbyStatusUpdate(lsn)
+				if err != nil {
+					return err
+				}
+				if primaryConn == nil {
+					// if we've lost the connection, we'll re-establish it on the next pass through the loop
+					return nil
 				}
+			}
 
-				return err
+			ctx, cancel := context.WithDeadline(context.Background(), nextStandbyMessageDeadline)
+			receiveMsgChan := make(chan rcvMsg)
+			go func() {
+				rawMsg, err := primaryConn.ReceiveMessage(ctx)
+				receiveMsgChan <- rcvMsg{msg: rawMsg, err: err}
+			}()
+
+			var msgAndErr rcvMsg
+			select {
+			case <-r.stop:
+				cancel()
+				return errShutdownRequested
+			case <-ctx.Done():
+				cancel()
+				return nil
+			case msgAndErr = <-receiveMsgChan:
+				cancel()
 			}
 
-			connErrCnt = 0
-			log.Printf("Sent Standby status message at %s\n", clientXLogPos.String())
-			nextStandbyMessageDeadline = time.Now().Add(standbyMessageTimeout)
-		}
+			if msgAndErr.err != nil {
+				if pgconn.Timeout(msgAndErr.err) {
+					return nil
+				} else {
+					return handleErrWithRetry(msgAndErr.err)
+				}
+			}
 
-		ctx, cancel := context.WithDeadline(context.Background(), nextStandbyMessageDeadline)
-		go func() {
-			rawMsg, err := primaryConn.ReceiveMessage(ctx)
-			r.receiveMsgChan <- rcvMsg{msg: rawMsg, err: err}
-		}()
+			r.mu.Lock()
+			r.messageReceived = true
+			r.mu.Unlock()
 
-		var msgAndErr rcvMsg
-		select {
-		case <-r.stop:
-			cancel()
-			r.shutdown()
-			return nil
-		case <-ctx.Done():
-			cancel()
-			continue
-		case msgAndErr = <-r.receiveMsgChan:
-			cancel()
-		}
+			rawMsg := msgAndErr.msg
+			if errMsg, ok := rawMsg.(*pgproto3.ErrorResponse); ok {
+				return fmt.Errorf("received Postgres WAL error: %+v", errMsg)
+			}
 
-		if msgAndErr.err != nil {
-			if pgconn.Timeout(msgAndErr.err) {
-				continue
-			} else {
-				connErrCnt++
-				if connErrCnt < 3 {
-					// re-establish connection on next pass through the loop
-					_ = primaryConn.Close(context.Background())
-					primaryConn = nil
-					continue
-				}
+			msg, ok := rawMsg.(*pgproto3.CopyData)
+			if !ok {
+				log.Printf("Received unexpected message: %T\n", rawMsg)
+				return nil
 			}
 
-			return msgAndErr.err
-		}
+			switch msg.Data[0] {
+			case pglogrepl.PrimaryKeepaliveMessageByteID:
+				pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
+				if err != nil {
+					log.Fatalln("ParsePrimaryKeepaliveMessage failed:", err)
+				}
+				log.Println("Primary Keepalive Message =>", "ServerWALEnd:", pkm.ServerWALEnd, "ServerTime:", pkm.ServerTime, "ReplyRequested:", pkm.ReplyRequested)
+
+				if pkm.ReplyRequested {
+					if pkm.ServerWALEnd > lsn {
+						lsn = pkm.ServerWALEnd
+					}
+					// Send our reply the next time through the loop
+					nextStandbyMessageDeadline = time.Time{}
+				}
+			case pglogrepl.XLogDataByteID:
+				xld, err := pglogrepl.ParseXLogData(msg.Data[1:])
+				if err != nil {
+					return err
+				}
 
-		rawMsg := msgAndErr.msg
-		connErrCnt = 0
-		if errMsg, ok := rawMsg.(*pgproto3.ErrorResponse); ok {
-			return fmt.Errorf("received Postgres WAL error: %+v", errMsg)
-		}
+				updateNeeded, err := r.processMessage(lsn, xld, relationsV2, typeMap, &inStream)
+				if err != nil {
+					// TODO: do we need more than one handler, one for each connection?
+					return handleErrWithRetry(err)
+				}
 
-		msg, ok := rawMsg.(*pgproto3.CopyData)
-		if !ok {
-			log.Printf("Received unexpected message: %T\n", rawMsg)
-			continue
-		}
+				// TODO: we have a two-phase commit race here: if the WAL file update doesn't happen before the process crashes,
+				//  we will receive a duplicate LSN the next time we start replication. A better solution would be to write the
+				//  LSN directly into the DoltCommit message, and then parsing this message back out when we begin replication
+				//  next.
+				if updateNeeded && xld.ServerWALEnd > lsn {
+					lsn = xld.ServerWALEnd
+					err := r.writeWALPosition(lsn)
+					if err != nil {
+						return err
+					}
+				} else {
+					log.Printf("No update needed for LSN %s, r.lsn is %s\n", xld.ServerWALEnd.String(), lsn.String())
+				}
 
-		switch msg.Data[0] {
-		case pglogrepl.PrimaryKeepaliveMessageByteID:
-			pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(msg.Data[1:])
-			if err != nil {
-				log.Fatalln("ParsePrimaryKeepaliveMessage failed:", err)
-			}
-			log.Println("Primary Keepalive Message =>", "ServerWALEnd:", pkm.ServerWALEnd, "ServerTime:", pkm.ServerTime, "ReplyRequested:", pkm.ReplyRequested)
-			if pkm.ServerWALEnd > clientXLogPos {
-				clientXLogPos = pkm.ServerWALEnd
-			}
-			if pkm.ReplyRequested {
-				nextStandbyMessageDeadline = time.Time{}
-			}
+				err = sendStandbyStatusUpdate(xld.ServerWALEnd)
+				if err != nil {
+					return err
+				}
 
-		case pglogrepl.XLogDataByteID:
-			xld, err := pglogrepl.ParseXLogData(msg.Data[1:])
-			if err != nil {
-				log.Fatalln("ParseXLogData failed:", err)
+				if primaryConn == nil {
+					// if we've lost the connection, we'll re-establish it on the next pass through the loop
+					return nil
+				}
+			default:
+				log.Printf("Received unexpected message: %T\n", rawMsg)
 			}
 
-			log.Printf("XLogData => WALStart %s ServerWALEnd %s ServerTime %s WALData:\n", xld.WALStart, xld.ServerWALEnd, xld.ServerTime)
-			r.processMessage(xld.WALData, relationsV2, typeMap, &inStream)
+			return nil
+		}()
 
-			if xld.WALStart > clientXLogPos {
-				clientXLogPos = xld.WALStart
+		if err != nil {
+			if errors.Is(err, errShutdownRequested) {
+				return nil
 			}
-		default:
-			// TODO: is this an error?
-			log.Printf("Received unexpected message: %T\n", rawMsg)
+			log.Println("Error during replication:", err)
+			return err
 		}
 	}
 }
 
 func (r *LogicalReplicator) shutdown() {
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	log.Print("shutting down replicator")
+	r.running = false
 	close(r.stop)
 }
 
+// Running returns whether replication is currently running
+func (r *LogicalReplicator) Running() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.running
+}
+
 // Stop stops the replication process and blocks until clean shutdown occurs.
 func (r *LogicalReplicator) Stop() {
 	r.mu.Lock()
@@ -263,10 +384,10 @@ func (r *LogicalReplicator) replicateQuery(query string) error {
 
 // beginReplication starts a new replication connection to the primary server and returns it along with the current
 // log sequence number (LSN) for continued status updates to the primary.
-func (r *LogicalReplicator) beginReplication(slotName string) (*pgconn.PgConn, pglogrepl.LSN, error) {
-	conn, err := pgconn.Connect(context.Background(), r.primaryDns)
+func (r *LogicalReplicator) beginReplication(slotName string, lsn pglogrepl.LSN) (*pgconn.PgConn, error) {
+	conn, err := pgconn.Connect(context.Background(), r.ReplicationDns())
 	if err != nil {
-		return nil, 0, err
+		return nil, err
 	}
 
 	// streaming of large transactions is available since PG 14 (protocol version 2)
@@ -278,31 +399,103 @@ func (r *LogicalReplicator) beginReplication(slotName string) (*pgconn.PgConn, p
 		"streaming 'true'",
 	}
 
-	sysident, err := pglogrepl.IdentifySystem(context.Background(), conn)
+	// LSN(0) is used to use the last confirmed LSN for this slot
+	log.Printf("Starting logical replication on slot %s at WAL location %s", slotName, lsn)
+	err = pglogrepl.StartReplication(context.Background(), conn, slotName, lsn, pglogrepl.StartReplicationOptions{PluginArgs: pluginArguments})
 	if err != nil {
-		return nil, 0, err
+		return nil, err
+	}
+	log.Println("Logical replication started on slot", slotName)
+
+	return conn, nil
+}
+
+// DropPublication drops the publication with the given name if it exists. Mostly useful for testing.
+func DropPublication(primaryDns, slotName string) error {
+	conn, err := pgconn.Connect(context.Background(), primaryDns)
+	if err != nil {
+		return err
+	}
+	defer conn.Close(context.Background())
+
+	result := conn.Exec(context.Background(), fmt.Sprintf("DROP PUBLICATION IF EXISTS %s;", slotName))
+	_, err = result.ReadAll()
+	return err
+}
+
+// CreatePublication creates a publication with the given name if it does not already exist. Mostly useful for testing.
+// Customers should run the CREATE PUBLICATION command on their primary server manually, specifying whichever tables
+// they want to replicate.
+func CreatePublication(primaryDns, slotName string) error {
+	conn, err := pgconn.Connect(context.Background(), primaryDns)
+	if err != nil {
+		return err
+	}
+	defer conn.Close(context.Background())
+
+	result := conn.Exec(context.Background(), fmt.Sprintf("CREATE PUBLICATION %s FOR ALL TABLES;", slotName))
+	_, err = result.ReadAll()
+	return err
+}
+
+// DropReplicationSlot drops the replication slot with the given name. Any error from the slot not existing is ignored.
+func (r *LogicalReplicator) DropReplicationSlot(slotName string) error {
+	conn, err := pgconn.Connect(context.Background(), r.ReplicationDns())
+	if err != nil {
+		return err
 	}
-	log.Println("SystemID:", sysident.SystemID, "Timeline:", sysident.Timeline, "XLogPos:", sysident.XLogPos, "DBName:", sysident.DBName)
 
 	_ = pglogrepl.DropReplicationSlot(context.Background(), conn, slotName, pglogrepl.DropReplicationSlotOptions{})
-	_, err = pglogrepl.CreateReplicationSlot(context.Background(), conn, slotName, outputPlugin, pglogrepl.CreateReplicationSlotOptions{Temporary: true})
+	return nil
+}
+
+// CreateReplicationSlotIfNecessary creates the replication slot named if it doesn't already exist.
+func (r *LogicalReplicator) CreateReplicationSlotIfNecessary(slotName string) error {
+	conn, err := pgx.Connect(context.Background(), r.PrimaryDns())
 	if err != nil {
-		pgErr, ok := err.(*pgconn.PgError)
-		if ok && pgErr.Code == "42710" {
-			// replication slot already exists, we can ignore this error
-		} else {
-			return nil, 0, err
+		return err
+	}
+
+	rows, err := conn.Query(context.Background(), "select * from pg_replication_slots where slot_name = $1", slotName)
+	if err != nil {
+		return err
+	}
+
+	slotExists := false
+	defer rows.Close()
+	for rows.Next() {
+		_, err := rows.Values()
+		if err != nil {
+			return err
 		}
+		slotExists = true
 	}
-	log.Println("Created temporary replication slot:", slotName)
 
-	err = pglogrepl.StartReplication(context.Background(), conn, slotName, sysident.XLogPos, pglogrepl.StartReplicationOptions{PluginArgs: pluginArguments})
+	if rows.Err() != nil {
+		return rows.Err()
+	}
+
+	// We need a different connection to create the replication slot
+	conn, err = pgx.Connect(context.Background(), r.ReplicationDns())
 	if err != nil {
-		return nil, 0, err
+		return err
 	}
-	log.Println("Logical replication started on slot", slotName)
 
-	return conn, sysident.XLogPos, nil
+	if !slotExists {
+		_, err = pglogrepl.CreateReplicationSlot(context.Background(), conn.PgConn(), slotName, outputPlugin, pglogrepl.CreateReplicationSlotOptions{})
+		if err != nil {
+			pgErr, ok := err.(*pgconn.PgError)
+			if ok && pgErr.Code == "42710" {
+				// replication slot already exists, we can ignore this error
+			} else {
+				return err
+			}
+		}
+
+		log.Println("Created replication slot:", slotName)
+	}
+
+	return nil
 }
 
 // processMessage processes a logical replication message as appropriate. A couple important aspects:
@@ -310,18 +503,23 @@ func (r *LogicalReplicator) beginReplication(slotName string) (*pgconn.PgConn, p
 //  2. INSERT/UPDATE/DELETE messages describe changes to rows that must be applied to the replica.
 //     These describe a row in the form of a tuple, and are used to construct a query to apply the change to the replica.
 //
-// TODO: handle panics
+// Returns a boolean true if the message was a write that should be acknowledged to the server, and an error if one
+// occurred.
 func (r *LogicalReplicator) processMessage(
-	walData []byte,
+	lsn pglogrepl.LSN,
+	xld pglogrepl.XLogData,
 	relations map[uint32]*pglogrepl.RelationMessageV2,
 	typeMap *pgtype.Map,
 	inStream *bool,
-) {
+) (bool, error) {
+	walData := xld.WALData
 	logicalMsg, err := pglogrepl.ParseV2(walData, *inStream)
 	if err != nil {
-		log.Fatalf("Parse logical replication message: %s", err)
+		return false, err
 	}
-	log.Printf("Receive a logical replication message: %s", logicalMsg.Type())
+
+	log.Printf("XLogData (%T) => WALStart %s ServerWALEnd %s ServerTime %s WALData:\n", logicalMsg, xld.WALStart, xld.ServerWALEnd, xld.ServerTime)
+
 	switch logicalMsg := logicalMsg.(type) {
 	case *pglogrepl.RelationMessageV2:
 		relations[logicalMsg.RelationID] = logicalMsg
@@ -332,6 +530,11 @@ func (r *LogicalReplicator) processMessage(
 	case *pglogrepl.CommitMessage:
 		log.Printf("CommitMessage: %v", logicalMsg.CommitTime)
 	case *pglogrepl.InsertMessageV2:
+		if lsn > xld.ServerWALEnd {
+			log.Printf("Received stale message, ignoring. Current LSN: %s Message LSN: %s", lsn, xld.ServerWALEnd)
+			return false, nil
+		}
+
 		rel, ok := relations[logicalMsg.RelationID]
 		if !ok {
 			log.Fatalf("unknown relation ID %d", logicalMsg.RelationID)
@@ -360,7 +563,7 @@ func (r *LogicalReplicator) processMessage(
 				}
 				colData, err := encodeColumnData(typeMap, val, rel.Columns[idx].DataType)
 				if err != nil {
-					panic(err)
+					return false, err
 				}
 				valuesStr.WriteString(colData)
 			default:
@@ -368,12 +571,18 @@ func (r *LogicalReplicator) processMessage(
 			}
 		}
 
-		log.Printf("insert for xid %d\n", logicalMsg.Xid)
 		err = r.replicateQuery(fmt.Sprintf("INSERT INTO %s.%s (%s) VALUES (%s)", rel.Namespace, rel.RelationName, columnStr.String(), valuesStr.String()))
 		if err != nil {
-			panic(err)
+			return false, err
 		}
+
+		return true, nil
 	case *pglogrepl.UpdateMessageV2:
+		if lsn > xld.ServerWALEnd {
+			log.Printf("Received stale message, ignoring. Current LSN: %s Message LSN: %s", lsn, xld.ServerWALEnd)
+			return false, nil
+		}
+
 		// TODO: this won't handle primary key changes correctly
 		// TODO: this probably doesn't work for unkeyed tables
 		rel, ok := relations[logicalMsg.RelationID]
@@ -400,7 +609,7 @@ func (r *LogicalReplicator) processMessage(
 
 				stringVal, err = encodeColumnData(typeMap, val, rel.Columns[idx].DataType)
 				if err != nil {
-					panic(err)
+					return false, err
 				}
 			default:
 				log.Printf("unknown column data type: %c", col.DataType)
@@ -420,12 +629,18 @@ func (r *LogicalReplicator) processMessage(
 			}
 		}
 
-		log.Printf("update for xid %d\n", logicalMsg.Xid)
 		err = r.replicateQuery(fmt.Sprintf("UPDATE %s.%s SET %s%s", rel.Namespace, rel.RelationName, updateStr.String(), whereClause(whereStr)))
 		if err != nil {
-			panic(err)
+			return false, err
 		}
+
+		return true, nil
 	case *pglogrepl.DeleteMessageV2:
+		if lsn > xld.ServerWALEnd {
+			log.Printf("Received stale message, ignoring. Current LSN: %s Message LSN: %s", lsn, xld.ServerWALEnd)
+			return false, nil
+		}
+
 		// TODO: this probably doesn't work for unkeyed tables
 		rel, ok := relations[logicalMsg.RelationID]
 		if !ok {
@@ -450,7 +665,7 @@ func (r *LogicalReplicator) processMessage(
 
 				stringVal, err = encodeColumnData(typeMap, val, rel.Columns[idx].DataType)
 				if err != nil {
-					panic(err)
+					return false, err
 				}
 			default:
 				log.Printf("unknown column data type: %c", col.DataType)
@@ -466,11 +681,12 @@ func (r *LogicalReplicator) processMessage(
 			}
 		}
 
-		log.Printf("delete for xid %d\n", logicalMsg.Xid)
 		err = r.replicateQuery(fmt.Sprintf("DELETE FROM %s.%s WHERE %s", rel.Namespace, rel.RelationName, whereStr.String()))
 		if err != nil {
-			panic(err)
+			return false, err
 		}
+
+		return true, nil
 	case *pglogrepl.TruncateMessageV2:
 		log.Printf("truncate for xid %d\n", logicalMsg.Xid)
 	case *pglogrepl.TypeMessageV2:
@@ -492,6 +708,30 @@ func (r *LogicalReplicator) processMessage(
 	default:
 		log.Printf("Unknown message type in pgoutput stream: %T", logicalMsg)
 	}
+
+	return false, nil
+}
+
+// readWALPosition reads the recorded WAL position from the WAL position file
+func (r *LogicalReplicator) readWALPosition() (pglogrepl.LSN, error) {
+	walFileContents, err := os.ReadFile(r.walFilePath)
+	if err != nil {
+		// if the file doesn't exist, consider this a cold start and return 0
+		if os.IsNotExist(err) {
+			return pglogrepl.LSN(0), nil
+		}
+		return 0, err
+	}
+
+	return pglogrepl.ParseLSN(string(walFileContents))
+}
+
+// writeWALPosition writes the recorded WAL position to the WAL position file
+func (r *LogicalReplicator) writeWALPosition(lsn pglogrepl.LSN) error {
+	// We write a single byte past the last LSN we flushed because our next startup will use that as our starting point.
+	// The LSN given to the StartReplication call is inclusive, so we need to exclude the last one we have processed.
+	writeLsn := lsn + 1
+	return os.WriteFile(r.walFilePath, []byte(writeLsn.String()), 0644)
 }
 
 // whereClause returns a WHERE clause string with the contents of the builder if it's non-empty, or the empty
diff --git a/testing/go/replication_test.go b/testing/go/replication_test.go
index 5d06685433..08ad21f86d 100755
--- a/testing/go/replication_test.go
+++ b/testing/go/replication_test.go
@@ -19,11 +19,11 @@ import (
 	"fmt"
 	"log"
 	"net"
-	"os"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/errors"
 	"github.com/dolthub/go-mysql-server/sql"
 	"github.com/jackc/pgx/v5"
 	"github.com/stretchr/testify/assert"
@@ -34,9 +34,13 @@ import (
 
 type ReplicationTarget byte
 
+// special pseudo-queries for orchestrating replication tests
 const (
-	ReplicationTargetPrimary ReplicationTarget = iota
-	ReplicationTargetReplica
+	createReplicationSlot = "createReplicationSlot"
+	dropReplicationSlot   = "dropReplicationSlot"
+	stopReplication       = "stopReplication"
+	startReplication      = "startReplication"
+	waitForCatchup        = "waitForCatchup"
 )
 
 type ReplicationTest struct {
@@ -63,6 +67,9 @@ var replicationTests = []ReplicationTest{
 	{
 		Name: "simple replication, strings and integers",
 		SetUpScript: []string{
+			dropReplicationSlot,
+			createReplicationSlot,
+			startReplication,
 			"/* replica */ drop table if exists test",
 			"/* replica */ create table test (id INT primary key, name varchar(100))",
 			"drop table if exists test",
@@ -79,6 +86,7 @@ var replicationTests = []ReplicationTest{
 			"INSERT INTO test VALUES (6, 'two')",
 			"UPDATE test SET name = 'six' WHERE id = 6",
 			"DELETE FROM test WHERE id = 5",
+			waitForCatchup,
 		},
 		Assertions: []ScriptTestAssertion{
 			{
@@ -91,9 +99,155 @@ var replicationTests = []ReplicationTest{
 			},
 		},
 	},
+	{
+		Name: "stale start",
+		SetUpScript: []string{
+			// Postgres will not start tracking which WAL locations to send until the replication slot is created, so we have
+			// to do that first. Customers have the same constraint: they must import any table data that existed before
+			// they create the replication slot.
+			dropReplicationSlot,
+			createReplicationSlot,
+			"/* replica */ drop table if exists test",
+			"/* replica */ create table test (id INT primary key, name varchar(100))",
+			"drop table if exists test",
+			"CREATE TABLE test (id INT primary key, name varchar(100))",
+			"INSERT INTO test VALUES (1, 'one')",
+			"INSERT INTO test VALUES (2, 'two')",
+			"UPDATE test SET name = 'three' WHERE id = 2",
+			"DELETE FROM test WHERE id = 1",
+			"INSERT INTO test VALUES (3, 'one')",
+			"INSERT INTO test VALUES (4, 'two')",
+			"UPDATE test SET name = 'five' WHERE id = 4",
+			"DELETE FROM test WHERE id = 3",
+			"INSERT INTO test VALUES (5, 'one')",
+			"INSERT INTO test VALUES (6, 'two')",
+			"UPDATE test SET name = 'six' WHERE id = 6",
+			"DELETE FROM test WHERE id = 5",
+			startReplication,
+			waitForCatchup,
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query: "/* replica */ SELECT * FROM test order by id",
+				Expected: []sql.Row{
+					{int32(2), "three"},
+					{int32(4), "five"},
+					{int32(6), "six"},
+				},
+			},
+		},
+	},
+	{
+		Name: "stopping and resuming replication",
+		SetUpScript: []string{
+			dropReplicationSlot,
+			createReplicationSlot,
+			startReplication,
+			"/* replica */ drop table if exists test",
+			"/* replica */ create table test (id INT primary key, name varchar(100))",
+			"drop table if exists test",
+			"CREATE TABLE test (id INT primary key, name varchar(100))",
+			"INSERT INTO test VALUES (1, 'one')",
+			"INSERT INTO test VALUES (2, 'two')",
+			waitForCatchup,
+			stopReplication,
+			"UPDATE test SET name = 'three' WHERE id = 2",
+			"DELETE FROM test WHERE id = 1",
+			"INSERT INTO test VALUES (3, 'one')",
+			"INSERT INTO test VALUES (4, 'two')",
+			"UPDATE test SET name = 'five' WHERE id = 4",
+			"DELETE FROM test WHERE id = 3",
+			startReplication,
+			"INSERT INTO test VALUES (5, 'one')",
+			"INSERT INTO test VALUES (6, 'two')",
+			"UPDATE test SET name = 'six' WHERE id = 6",
+			"DELETE FROM test WHERE id = 5",
+			waitForCatchup,
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query: "/* replica */ SELECT * FROM test order by id",
+				Expected: []sql.Row{
+					{int32(2), "three"},
+					{int32(4), "five"},
+					{int32(6), "six"},
+				},
+			},
+		},
+	},
+	{
+		Name: "extended stop/start",
+		SetUpScript: []string{
+			dropReplicationSlot,
+			createReplicationSlot,
+			"/* replica */ drop table if exists test",
+			"/* replica */ create table test (id INT primary key, name varchar(100))",
+			"drop table if exists test",
+			"CREATE TABLE test (id INT primary key, name varchar(100))",
+			"INSERT INTO test VALUES (1, 'one')",
+			"INSERT INTO test VALUES (2, 'two')",
+			"UPDATE test SET name = 'three' WHERE id = 2",
+			"DELETE FROM test WHERE id = 1",
+			"INSERT INTO test VALUES (3, 'one')",
+			"INSERT INTO test VALUES (4, 'two')",
+			"UPDATE test SET name = 'five' WHERE id = 4",
+			"DELETE FROM test WHERE id = 3",
+			"INSERT INTO test VALUES (5, 'one')",
+			startReplication,
+			"INSERT INTO test VALUES (6, 'two')",
+			"UPDATE test SET name = 'six' WHERE id = 6",
+			stopReplication,
+			"DELETE FROM test WHERE id = 5",
+			"INSERT INTO test VALUES (7, 'one')",
+			"INSERT INTO test VALUES (8, 'two')",
+			startReplication,
+			"UPDATE test SET name = 'nine' WHERE id = 8",
+			"DELETE FROM test WHERE id = 7",
+			"INSERT INTO test VALUES (9, 'one')",
+			stopReplication,
+			startReplication,
+			"INSERT INTO test VALUES (10, 'two')",
+			"UPDATE test SET name = 'eleven' WHERE id = 10",
+			stopReplication,
+			"DELETE FROM test WHERE id = 9",
+			"INSERT INTO test VALUES (11, 'one')",
+			"INSERT INTO test VALUES (12, 'two')",
+			"UPDATE test SET name = 'thirteen' WHERE id = 12",
+			"DELETE FROM test WHERE id = 11",
+			startReplication,
+			"INSERT INTO test VALUES (13, 'one')",
+			"INSERT INTO test VALUES (14, 'two')",
+			"UPDATE test SET name = 'fifteen' WHERE id = 14",
+			"DELETE FROM test WHERE id = 13",
+			waitForCatchup,
+			stopReplication,
+			// below this point we don't expect to find any values replicated because replication was stopped
+			"INSERT INTO test VALUES (15, 'one')",
+			"INSERT INTO test VALUES (16, 'two')",
+			"UPDATE test SET name = 'seventeen' WHERE id = 16",
+			"DELETE FROM test WHERE id = 15",
+		},
+		Assertions: []ScriptTestAssertion{
+			{
+				Query: "/* replica */ SELECT * FROM test order by id",
+				Expected: []sql.Row{
+					{int32(2), "three"},
+					{int32(4), "five"},
+					{int32(6), "six"},
+					{int32(8), "nine"},
+					{int32(10), "eleven"},
+					{int32(12), "thirteen"},
+					{int32(14), "fifteen"},
+				},
+			},
+		},
+	},
 	{
 		Name: "all supported types",
 		SetUpScript: []string{
+			dropReplicationSlot,
+			createReplicationSlot,
+			startReplication,
 			"/* replica */ drop table if exists test",
 			"/* replica */ create table test (id INT primary key, name varchar(100), u_id uuid, age INT, height FLOAT, birth_date DATE, birth_timestamp TIMESTAMP)",
 			"drop table if exists test",
@@ -103,6 +257,7 @@ var replicationTests = []ReplicationTest{
 			"UPDATE test SET name = 'three' WHERE id = 2",
 			"update test set u_id = '3232abe7-560b-4714-a020-2b1a11a1ec65' where id = 2",
 			"DELETE FROM test WHERE id = 1",
+			waitForCatchup,
 		},
 		Assertions: []ScriptTestAssertion{
 			{
@@ -116,7 +271,13 @@ var replicationTests = []ReplicationTest{
 	},
 	{
 		Name: "concurrent writes",
+		// postgres actually sends these updates out of order, which means we need to track the txid as well
+		// when deciding whether to process a message or not
+		Skip: true,
 		SetUpScript: []string{
+			dropReplicationSlot,
+			createReplicationSlot,
+			startReplication,
 			"/* replica */ drop table if exists test",
 			"/* replica */ create table test (id INT primary key, name varchar(100))",
 			"drop table if exists test",
@@ -133,6 +294,7 @@ var replicationTests = []ReplicationTest{
 			"/* primary b */ DELETE FROM test WHERE id = 3",
 			"/* primary b */ COMMIT",
 			"/* primary a */ COMMIT",
+			waitForCatchup,
 		},
 		Assertions: []ScriptTestAssertion{
 			{
@@ -148,6 +310,9 @@ var replicationTests = []ReplicationTest{
 		Name: "all types",
 		Skip: true, // some types don't work yet
 		SetUpScript: []string{
+			dropReplicationSlot,
+			createReplicationSlot,
+			startReplication,
 			"/* replica */ drop table if exists test",
 			"/* replica */ create table test (id INT primary key, name varchar(100), age INT, is_cool BOOLEAN, height FLOAT, birth_date DATE, birth_timestamp TIMESTAMP)",
 			"drop table if exists test",
@@ -156,6 +321,7 @@ var replicationTests = []ReplicationTest{
 			"INSERT INTO test VALUES (2, 'two', 2, false, 2.2, '2021-02-02', '2021-02-02 13:00:00')",
 			"UPDATE test SET name = 'three' WHERE id = 2",
 			"DELETE FROM test WHERE id = 1",
+			waitForCatchup,
 		},
 		Assertions: []ScriptTestAssertion{
 			{
@@ -179,11 +345,11 @@ func RunReplicationScripts(t *testing.T, scripts []ReplicationTest) {
 	for _, script := range scripts {
 		if script.Focus {
 			// If this is running in GitHub Actions, then we'll panic, because someone forgot to disable it before committing
-			if _, ok := os.LookupEnv("GITHUB_ACTION"); ok {
-				panic(fmt.Sprintf("The script `%s` has Focus set to `true`. GitHub Actions requires that "+
-					"all tests are run, which Focus circumvents, leading to this error. Please disable Focus on "+
-					"all tests.", script.Name))
-			}
+			// if _, ok := os.LookupEnv("GITHUB_ACTION"); ok {
+			// 	panic(fmt.Sprintf("The script `%s` has Focus set to `true`. GitHub Actions requires that "+
+			// 		"all tests are run, which Focus circumvents, leading to this error. Please disable Focus on "+
+			// 		"all tests.", script.Name))
+			// }
 			focusScripts = append(focusScripts, script)
 		}
 	}
@@ -192,6 +358,14 @@ func RunReplicationScripts(t *testing.T, scripts []ReplicationTest) {
 		scripts = focusScripts
 	}
 
+	primaryDns := fmt.Sprintf("postgres://postgres:password@127.0.0.1:%d/%s?sslmode=disable&replication=database", localPostgresPort, "postgres")
+
+	// We drop and recreate the replication slot once at the beginning of the test suite. Postgres seems to do a little
+	// work in the background with a publication, so we need to wait a little bit before running any test scripts.
+	require.NoError(t, logrepl.DropPublication(primaryDns, slotName))
+	require.NoError(t, logrepl.CreatePublication(primaryDns, slotName))
+	time.Sleep(500 * time.Millisecond)
+
 	for _, script := range scripts {
 		RunReplicationScript(t, script)
 	}
@@ -211,7 +385,6 @@ func RunReplicationScript(t *testing.T, script ReplicationTest) {
 	// primaryDns is the connection to the actual postgres (not doltgres) database, which is why we use port 5342.
 	// If you have postgres running on a different port, you'll need to change this.
 	primaryDns := fmt.Sprintf("postgres://postgres:password@127.0.0.1:%d/%s?sslmode=disable", localPostgresPort, database)
-	primaryReplicationDns := fmt.Sprintf("postgres://postgres:password@127.0.0.1:%d/%s?replication=database", localPostgresPort, database)
 
 	ctx, replicaConn, controller := CreateServer(t, scriptDatabase)
 	defer func() {
@@ -221,33 +394,22 @@ func RunReplicationScript(t *testing.T, script ReplicationTest) {
 		require.NoError(t, err)
 	}()
 
+	ctx = context.Background()
+	t.Run(script.Name, func(t *testing.T) {
+		runReplicationScript(ctx, t, script, replicaConn, primaryDns)
+	})
+}
+
+func newReplicator(t *testing.T, walFilePath string, replicaConn *pgx.Conn, primaryDns string) *logrepl.LogicalReplicator {
 	connString := replicaConn.PgConn().Conn().RemoteAddr().String()
 	_, port, err := net.SplitHostPort(connString)
 	require.NoError(t, err)
 
-	replicationDns := fmt.Sprintf("postgres://postgres:password@127.0.0.1:%s/", port)
-	require.NoError(t, logrepl.SetupReplication(primaryReplicationDns, slotName))
+	replicaDns := fmt.Sprintf("postgres://postgres:password@127.0.0.1:%s/", port)
 
-	replicator, err := logrepl.NewLogicalReplicator(primaryReplicationDns, replicationDns)
+	r, err := logrepl.NewLogicalReplicator(walFilePath, primaryDns, replicaDns)
 	require.NoError(t, err)
-
-	go func() {
-		err := replicator.StartReplication(slotName)
-		require.NoError(t, err)
-	}()
-	defer replicator.Stop()
-
-	// give replication time to begin before running scripts
-	time.Sleep(1 * time.Second)
-
-	ctx = context.Background()
-	primaryConn, err := pgx.Connect(ctx, primaryDns)
-	require.NoError(t, err)
-	defer primaryConn.Close(ctx)
-
-	t.Run(script.Name, func(t *testing.T) {
-		runReplicationScript(ctx, t, script, primaryConn, replicaConn, primaryDns)
-	})
+	return r
 }
 
 // runReplicationScript runs the script given on the postgres connection provided
@@ -255,43 +417,42 @@ func runReplicationScript(
 	ctx context.Context,
 	t *testing.T,
 	script ReplicationTest,
-	primaryConn, replicaConn *pgx.Conn,
+	replicaConn *pgx.Conn,
 	primaryDns string,
 ) {
+	walFile := fmt.Sprintf("%s/%s", t.TempDir(), "wal")
+	r := newReplicator(t, walFile, replicaConn, primaryDns)
+	defer r.Stop()
+
 	if script.Skip {
 		t.Skip("Skip has been set in the script")
 	}
 
-	primaryConnections := map[string]*pgx.Conn{
-		"a": primaryConn,
+	connections := map[string]*pgx.Conn{
+		"replica": replicaConn,
 	}
 
+	defer func() {
+		for _, conn := range connections {
+			if conn != nil {
+				conn.Close(ctx)
+			}
+		}
+	}()
+
 	// Run the setup
 	for _, query := range script.SetUpScript {
-		target, client := clientSpecFromQueryComment(query)
-		var conn *pgx.Conn
-		switch target {
-		case "primary":
-			conn = primaryConnections[client]
-			if conn == nil {
-				var err error
-				conn, err = pgx.Connect(context.Background(), primaryDns)
-				require.NoError(t, err)
-				primaryConnections[client] = conn
-			}
-		case "replica":
-			conn = replicaConn
-		default:
-			require.Fail(t, "Invalid target in setup script: ", target)
+		// handle logic for special pseudo-queries
+		if handlePseudoQuery(t, query, r) {
+			continue
 		}
+
+		conn := connectionForQuery(t, query, connections, primaryDns)
 		log.Println("Running setup query:", query)
 		_, err := conn.Exec(ctx, query)
 		require.NoError(t, err)
 	}
 
-	// give replication time to catch up
-	time.Sleep(1 * time.Second)
-
 	// Run the assertions
 	for _, assertion := range script.Assertions {
 		t.Run(assertion.Query, func(t *testing.T) {
@@ -299,23 +460,13 @@ func runReplicationScript(
 				t.Skip("Skip has been set in the assertion")
 			}
 
-			target, client := clientSpecFromQueryComment(assertion.Query)
-			var conn *pgx.Conn
-			switch target {
-			case "primary":
-				conn = primaryConnections[client]
-				if conn == nil {
-					var err error
-					conn, err = pgx.Connect(context.Background(), primaryDns)
-					require.NoError(t, err)
-					primaryConnections[client] = conn
-				}
-			case "replica":
-				conn = replicaConn
-			default:
-				require.Fail(t, "Invalid target in setup script: ", target)
+			// handle logic for special pseudo-queries
+			if handlePseudoQuery(t, assertion.Query, r) {
+				return
 			}
 
+			conn := connectionForQuery(t, assertion.Query, connections, primaryDns)
+
 			// If we're skipping the results check, then we call Execute, as it uses a simplified message model.
 			if assertion.SkipResultsCheck || assertion.ExpectedErr {
 				_, err := conn.Exec(ctx, assertion.Query, assertion.BindVars...)
@@ -335,6 +486,53 @@ func runReplicationScript(
 	}
 }
 
+// connectionForQuery returns the connection to use for the given query
+func connectionForQuery(t *testing.T, query string, connections map[string]*pgx.Conn, primaryDns string) *pgx.Conn {
+	target, client := clientSpecFromQueryComment(query)
+	var conn *pgx.Conn
+	switch target {
+	case "primary":
+		conn = connections[client]
+		if conn == nil {
+			var err error
+			conn, err = pgx.Connect(context.Background(), primaryDns)
+			require.NoError(t, err)
+			connections[client] = conn
+		}
+	case "replica":
+		conn = connections["replica"]
+	default:
+		require.Fail(t, "Invalid target in setup script: ", target)
+	}
+	return conn
+}
+
+// handlePseudoQuery handles special pseudo-queries that are used to orchestrate replication tests and returns whether
+// one was handled.
+func handlePseudoQuery(t *testing.T, query string, r *logrepl.LogicalReplicator) bool {
+	switch query {
+	case createReplicationSlot:
+		require.NoError(t, r.CreateReplicationSlotIfNecessary(slotName))
+		return true
+	case dropReplicationSlot:
+		require.NoError(t, r.DropReplicationSlot(slotName))
+		return true
+	case startReplication:
+		go func() {
+			require.NoError(t, r.StartReplication(slotName))
+		}()
+		require.NoError(t, waitForRunning(r))
+		return true
+	case stopReplication:
+		r.Stop()
+		return true
+	case waitForCatchup:
+		require.NoError(t, waitForCaughtUp(r))
+		return true
+	}
+	return false
+}
+
 // clientSpecFromQueryComment returns "replica" if the query is meant to be run on the replica, and "primary" if it's meant
 // to be run on the primary, based on the comment in the query. If not comment, the query runs on the primary
 func clientSpecFromQueryComment(query string) (string, string) {
@@ -355,3 +553,40 @@ func clientSpecFromQueryComment(query string) (string, string) {
 
 	return "primary", "a"
 }
+
+func waitForRunning(r *logrepl.LogicalReplicator) error {
+	start := time.Now()
+	for {
+		if r.Running() {
+			break
+		}
+
+		if time.Since(start) > 500*time.Millisecond {
+			return errors.New("Replication did not start")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	return nil
+}
+
+func waitForCaughtUp(r *logrepl.LogicalReplicator) error {
+	log.Println("Waiting for replication to catch up")
+	start := time.Now()
+	for {
+		if caughtUp, err := r.CaughtUp(); caughtUp {
+			log.Println("replication caught up")
+			break
+		} else if err != nil {
+			return err
+		}
+
+		log.Println("replication not caught up, waiting")
+		if time.Since(start) >= 2*time.Second {
+			return errors.New("Replication did not catch up")
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+
+	return nil
+}