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. This method uses a threshold value to determine if the primary considers us caught up. This co
(threshold int)
| 88 | // additional WAL locations cannot be recorded as flushed since they don't result in writes to the replica, and could |
| 89 | // result in the primary not sending us necessary records after a shutdown and restart. |
| 90 | func (r *LogicalReplicator) CaughtUp(threshold int) (bool, error) { |
| 91 | r.mu.Lock() |
| 92 | if !r.messageReceived { |
| 93 | r.mu.Unlock() |
| 94 | // We can't query the replication state until after receiving our first message |
| 95 | return false, nil |
| 96 | } |
| 97 | r.mu.Unlock() |
| 98 | |
| 99 | conn, err := pgx.Connect(context.Background(), r.PrimaryDns()) |
| 100 | if err != nil { |
| 101 | return false, err |
| 102 | } |
| 103 | defer conn.Close(context.Background()) |
| 104 | |
| 105 | result, err := conn.Query(context.Background(), "SELECT pg_wal_lsn_diff(write_lsn, sent_lsn) AS replication_lag FROM pg_stat_replication") |
| 106 | if err != nil { |
| 107 | return false, err |
| 108 | } |
| 109 | |
| 110 | defer result.Close() |
| 111 | |
| 112 | for result.Next() { |
| 113 | rows, err := result.Values() |
| 114 | if err != nil { |
| 115 | return false, err |
| 116 | } |
| 117 | |
| 118 | row := rows[0] |
| 119 | lag, ok := row.(pgtype.Numeric) |
| 120 | if ok && lag.Valid { |
| 121 | log.Printf("Current replication lag: %v", row) |
| 122 | return int(math.Abs(float64(lag.Int.Int64()))) < threshold, nil |
| 123 | } else { |
| 124 | log.Printf("Replication lag unknown: %v", row) |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | if result.Err() != nil { |
| 129 | return false, result.Err() |
| 130 | } |
| 131 | |
| 132 | // If we didn't get any rows, that usually means that replication has stopped and we're caught up |
| 133 | return true, nil |
| 134 | } |
| 135 | |
| 136 | // maxConsecutiveFailures is the maximum number of consecutive RPC errors that can occur before we stop |
| 137 | // the replication thread |