Implement the PostgreSQL Streaming Replication Protocol

.github/CODEOWNERS

@@ -1 +1,2 @@
-*       @roji @YohDeadfall
+*                        @roji @YohDeadfall
+/src/Npgsql/Replication/ @Brar

.github/workflows/build.yml

@@ -67,10 +67,16 @@ jobs:
           sudo sed -i 's/#ssl = off/ssl = on/' $PGDATA/postgresql.conf
           sudo sed -i 's/#max_prepared_transactions = 0/max_prepared_transactions = 10/' $PGDATA/postgresql.conf
           sudo sed -i 's/#password_encryption = md5/password_encryption = scram-sha-256/' $PGDATA/postgresql.conf
+          sudo sed -i 's/#wal_level =/wal_level = logical #/' $PGDATA/postgresql.conf
+          sudo sed -i 's/#max_wal_senders =/max_wal_senders = 50 #/' $PGDATA/postgresql.conf
+          sudo sed -i 's/#wal_sender_timeout =/wal_sender_timeout = 3s #/' $PGDATA/postgresql.conf
+          sudo sed -i "s/#synchronous_standby_names =/synchronous_standby_names = 'npgsql_test_sync_standby' #/" $PGDATA/postgresql.conf
+          sudo sed -i "s/#synchronous_commit =/synchronous_commit = local #/" $PGDATA/postgresql.conf
           # Disable trust authentication, requiring MD5 passwords - some tests must fail if a password isn't provided.
           sudo sh -c "echo 'local all all trust' > $PGDATA/pg_hba.conf"
           sudo sh -c "echo 'host all npgsql_tests_scram all scram-sha-256' >> $PGDATA/pg_hba.conf"
           sudo sh -c "echo 'host all all all md5' >> $PGDATA/pg_hba.conf"
+          sudo sh -c "echo 'host replication all all md5' >> $PGDATA/pg_hba.conf"
           sudo pg_ctlcluster ${{ matrix.pg_major }} main restart
 
           # user 'npgsql_tests_scram' must be created with password encrypted as scram-sha-256 (which only applies after restart)
@@ -114,6 +120,11 @@ jobs:
           pgsql/bin/initdb -D pgsql/PGDATA -E UTF8 -U postgres
           SOCKET_DIR=$(echo "$LOCALAPPDATA\Temp" | sed 's|\\|/|g')
           sed -i "s|#unix_socket_directories = ''|unix_socket_directories = '$SOCKET_DIR'|" pgsql/PGDATA/postgresql.conf
+          sed -i "s|#wal_level =|wal_level = logical #|" pgsql/PGDATA/postgresql.conf
+          sed -i "s|#max_wal_senders =|max_wal_senders = 50 #|" pgsql/PGDATA/postgresql.conf
+          sed -i "s|#wal_sender_timeout =|wal_sender_timeout = 3s #|" pgsql/PGDATA/postgresql.conf
+          sed -i "s|#synchronous_standby_names =|synchronous_standby_names = 'npgsql_test_sync_standby' #|" pgsql/PGDATA/postgresql.conf
+          sed -i "s|#synchronous_commit =|synchronous_commit = local #|" pgsql/PGDATA/postgresql.conf
           pgsql/bin/pg_ctl -D pgsql/PGDATA -l logfile -o '-c max_prepared_transactions=10 -c ssl=true -c ssl_cert_file=../server.crt -c ssl_key_file=../server.key' start
 
           # Configure test account
@@ -133,6 +144,7 @@ jobs:
           # Disable trust authentication, requiring MD5 passwords - some tests must fail if a password isn't provided.
           echo "host all npgsql_tests_scram all scram-sha-256" > pgsql/PGDATA/pg_hba.conf
           echo "host all all all md5" >> pgsql/PGDATA/pg_hba.conf
+          echo "host replication all all md5" >> pgsql/PGDATA/pg_hba.conf
 
       # TODO: Once test/Npgsql.Specification.Tests work, switch to just testing on the solution
       - name: Test

Directory.Build.props

@@ -15,7 +15,7 @@
     <NoWarn>$(NoWarn);NU5105</NoWarn>
 
     <!-- Language -->
-    <LangVersion>8.0</LangVersion>
+    <LangVersion>9.0</LangVersion>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <Nullable>enable</Nullable>
     <AnalysisLevel>latest</AnalysisLevel>

Npgsql.sln.DotSettings

@@ -88,6 +88,7 @@
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=MSDTC/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=multiquery/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Noda/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=NOEXPORT/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Npgsql/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Npgsql_0027s/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=PGTZ/@EntryIndexedValue">True</s:Boolean>

src/Npgsql/Common.cs

@@ -46,6 +46,7 @@ static class FrontendMessageCode
         internal const byte Bind =      (byte)'B';
         internal const byte Close =     (byte)'C';
         internal const byte Query =     (byte)'Q';
+        internal const byte CopyData =  (byte)'d';
         internal const byte CopyDone =  (byte)'c';
         internal const byte CopyFail =  (byte)'f';
         internal const byte Terminate = (byte)'X';

src/Npgsql/NpgsqlConnection.cs

@@ -462,14 +462,15 @@ public ConnectionState FullState
                         ? ConnectionState.Open // When unbound, we only know we're open
                         : Connector.State switch
                         {
-                            ConnectorState.Ready      => ConnectionState.Open,
-                            ConnectorState.Executing  => ConnectionState.Open | ConnectionState.Executing,
-                            ConnectorState.Fetching   => ConnectionState.Open | ConnectionState.Fetching,
-                            ConnectorState.Copy       => ConnectionState.Open | ConnectionState.Fetching,
-                            ConnectorState.Waiting    => ConnectionState.Open | ConnectionState.Fetching,
-                            ConnectorState.Connecting => ConnectionState.Connecting,
-                            ConnectorState.Broken     => ConnectionState.Broken,
-                            ConnectorState.Closed     => throw new InvalidOperationException("Internal Npgsql bug: connection is in state Open but connector is in state Closed"),
+                            ConnectorState.Ready       => ConnectionState.Open,
+                            ConnectorState.Executing   => ConnectionState.Open | ConnectionState.Executing,
+                            ConnectorState.Fetching    => ConnectionState.Open | ConnectionState.Fetching,
+                            ConnectorState.Copy        => ConnectionState.Open | ConnectionState.Fetching,
+                            ConnectorState.Replication => ConnectionState.Open | ConnectionState.Fetching,
+                            ConnectorState.Waiting     => ConnectionState.Open | ConnectionState.Fetching,
+                            ConnectorState.Connecting  => ConnectionState.Connecting,
+                            ConnectorState.Broken      => ConnectionState.Broken,
+                            ConnectorState.Closed      => throw new InvalidOperationException("Internal Npgsql bug: connection is in state Open but connector is in state Closed"),
                             _ => throw new InvalidOperationException($"Internal Npgsql bug: unexpected value {Connector.State} of enum {nameof(ConnectorState)}. Please file a bug.")
                         },
                     _ => _fullState

src/Npgsql/NpgsqlConnectionStringBuilder.cs

@@ -9,6 +9,7 @@
 using System.Linq;
 using System.Reflection;
 using System.Text;
+using Npgsql.Replication;
 
 namespace Npgsql
 {
@@ -89,7 +90,7 @@ void Init()
         static NpgsqlConnectionStringBuilder()
         {
             var properties = typeof(NpgsqlConnectionStringBuilder)
-                .GetProperties()
+                .GetProperties(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic)
                 .Where(p => p.GetCustomAttribute<NpgsqlConnectionStringPropertyAttribute>() != null)
                 .ToArray();
 
@@ -1252,6 +1253,30 @@ public bool LoadTableComposites
         }
         bool _loadTableComposites;
 
+        /// <summary>
+        /// Set the replication mode of the connection
+        /// </summary>
+        /// <remarks>
+        /// This property and its corresponding enum are intentionally kept internal as they
+        /// should not be set by users or even be visible in their connection strings.
+        /// Replication connections are a special kind of connection that is encapsulated in
+        /// <see cref="PhysicalReplicationConnection"/>
+        /// and <see cref="LogicalReplicationConnection"/>.
+        /// </remarks>
+
+        [NpgsqlConnectionStringProperty]
+        [DisplayName("Replication Mode")]
+        internal ReplicationMode ReplicationMode
+        {
+            get => _replicationMode;
+            set
+            {
+                _replicationMode = value;
+                SetValue(nameof(ReplicationMode), value);
+            }
+        }
+        ReplicationMode _replicationMode;
+
         /// <summary>
         /// Set PostgreSQL configuration parameter default values for the connection.
         /// </summary>
@@ -1698,5 +1723,31 @@ public enum SslMode
         Require,
     }
 
+    /// <summary>
+    /// Specifies whether the connection shall be initialized as a physical or
+    /// logical replication connection
+    /// </summary>
+    /// <remarks>
+    /// This enum and its corresponding property are intentionally kept internal as they
+    /// should not be set by users or even be visible in their connection strings.
+    /// Replication connections are a special kind of connection that is encapsulated in
+    /// <see cref="PhysicalReplicationConnection"/>
+    /// and <see cref="LogicalReplicationConnection"/>.
+    /// </remarks>
+    enum ReplicationMode
+    {
+        /// <summary>
+        /// Replication disabled. This is the default
+        /// </summary>
+        Off,
+        /// <summary>
+        /// Physical replication enabled
+        /// </summary>
+        Physical,
+        /// <summary>
+        /// Logical replication enabled
+        /// </summary>
+        Logical
+    }
     #endregion
 }

src/Npgsql/NpgsqlConnector.cs

@@ -261,6 +261,7 @@ internal void FlagAsWritableForMultiplexing()
         CopyInResponseMessage?  _copyInResponseMessage;
         CopyOutResponseMessage? _copyOutResponseMessage;
         CopyDataMessage?        _copyDataMessage;
+        CopyBothResponseMessage? _copyBothResponseMessage;
 
         #endregion
 
@@ -393,15 +394,16 @@ internal ConnectorState State
         bool IsConnected
             => State switch
             {
-                ConnectorState.Ready      => true,
-                ConnectorState.Executing  => true,
-                ConnectorState.Fetching   => true,
-                ConnectorState.Waiting    => true,
-                ConnectorState.Copy       => true,
-                ConnectorState.Closed     => false,
-                ConnectorState.Connecting => false,
-                ConnectorState.Broken     => false,
-                _                         => throw new ArgumentOutOfRangeException("Unknown state: " + State)
+                ConnectorState.Ready       => true,
+                ConnectorState.Executing   => true,
+                ConnectorState.Fetching    => true,
+                ConnectorState.Waiting     => true,
+                ConnectorState.Copy        => true,
+                ConnectorState.Replication => true,
+                ConnectorState.Closed      => false,
+                ConnectorState.Connecting  => false,
+                ConnectorState.Broken      => false,
+                _                          => throw new ArgumentOutOfRangeException("Unknown state: " + State)
             };
 
         internal bool IsReady => State == ConnectorState.Ready;
@@ -536,6 +538,16 @@ void WriteStartupMessage(string username)
             if (timezone != null)
                 startupParams["TimeZone"] = timezone;
 
+            switch (Settings.ReplicationMode)
+            {
+            case ReplicationMode.Logical:
+                startupParams["replication"] = "database";
+                break;
+            case ReplicationMode.Physical:
+                startupParams["replication"] = "true";
+                break;
+            }
+
             WriteStartup(startupParams);
         }
 
@@ -1281,6 +1293,9 @@ await ReadMessageLong(DataRowLoadingMode.Skip, readingNotifications2: false, isR
                     return (_copyOutResponseMessage ??= new CopyOutResponseMessage()).Load(ReadBuffer);
                 case BackendMessageCode.CopyData:
                     return (_copyDataMessage ??= new CopyDataMessage()).Load(len);
+                case BackendMessageCode.CopyBothResponse:
+                    return (_copyBothResponseMessage ??= new CopyBothResponseMessage()).Load(ReadBuffer);
+
                 case BackendMessageCode.CopyDone:
                     return CopyDoneMessage.Instance;
 
@@ -1834,7 +1849,7 @@ internal void UnprepareAll()
         internal UserAction StartUserAction(NpgsqlCommand command)
             => StartUserAction(ConnectorState.Executing, command);
 
-        internal UserAction StartUserAction(ConnectorState newState=ConnectorState.Executing, NpgsqlCommand? command = null)
+        internal UserAction StartUserAction(ConnectorState newState = ConnectorState.Executing, NpgsqlCommand? command = null)
         {
             // If keepalive is enabled, we must protect state transitions with a SemaphoreSlim
             // (which itself must be protected by a lock, since its dispose isn't thread-safe).
@@ -1884,6 +1899,7 @@ UserAction DoStartUserAction()
                 case ConnectorState.Executing:
                 case ConnectorState.Fetching:
                 case ConnectorState.Waiting:
+                case ConnectorState.Replication:
                 case ConnectorState.Connecting:
                 case ConnectorState.Copy:
                     var currentCommand = _currentCommand;
@@ -2205,6 +2221,11 @@ enum ConnectorState
         /// The connector is engaged in a COPY operation.
         /// </summary>
         Copy,
+
+        /// <summary>
+        /// The connector is engaged in streaming replication.
+        /// </summary>
+        Replication,
     }
 
 #pragma warning disable CA1717

src/Npgsql/NpgsqlDataReader.cs

@@ -882,6 +882,7 @@ internal async Task Close(bool connectionClosing, bool async)
                 break;
             case ConnectorState.Waiting:
             case ConnectorState.Copy:
+            case ConnectorState.Replication:
                 Debug.Fail("Bad connector state when closing reader: " + Connector.State);
                 break;
             default:

src/Npgsql/NpgsqlReadBuffer.cs

@@ -505,6 +505,9 @@ public int Read(Span<byte> output)
 
         public ValueTask<int> ReadAsync(Memory<byte> output, CancellationToken cancellationToken = default)
         {
+            if (output.Length == 0)
+                return new ValueTask<int>(0);
+
             var readFromBuffer = Math.Min(ReadBytesLeft, output.Length);
             if (readFromBuffer > 0)
             {
@@ -513,9 +516,6 @@ public ValueTask<int> ReadAsync(Memory<byte> output, CancellationToken cancellat
                 return new ValueTask<int>(readFromBuffer);
             }
 
-            if (output.Length == 0)
-                return new ValueTask<int>(0);
-
             return ReadAsyncLong();
 
             async ValueTask<int> ReadAsyncLong()
@@ -549,29 +549,60 @@ public Stream GetStream(int len, bool canSeek)
         /// Seeks the first null terminator (\0) and returns the string up to it. The buffer must already
         /// contain the entire string and its terminator.
         /// </summary>
-        public string ReadNullTerminatedString() => ReadNullTerminatedString(TextEncoding);
+        public string ReadNullTerminatedString()
+            => ReadNullTerminatedString(TextEncoding, async: false).GetAwaiter().GetResult();
 
         /// <summary>
         /// Seeks the first null terminator (\0) and returns the string up to it. The buffer must already
         /// contain the entire string and its terminator. If any character could not be decoded, a question
         /// mark character is returned instead of throwing an exception.
         /// </summary>
-        public string ReadNullTerminatedStringRelaxed() => ReadNullTerminatedString(RelaxedTextEncoding);
+        public string ReadNullTerminatedStringRelaxed()
+            => ReadNullTerminatedString(RelaxedTextEncoding, async: false).GetAwaiter().GetResult();
+
+        public ValueTask<string> ReadNullTerminatedString(bool async, CancellationToken cancellationToken = default)
+            => ReadNullTerminatedString(TextEncoding, async, cancellationToken);
 
         /// <summary>
-        /// Seeks the first null terminator (\0) and returns the string up to it. The buffer must already
-        /// contain the entire string and its terminator.
+        /// Seeks the first null terminator (\0) and returns the string up to it. Reads additional data from the network if a null
+        /// terminator isn't found in the buffered data.
         /// </summary>
-        /// <param name="encoding">Decodes the messages with this encoding.</param>
-        string ReadNullTerminatedString(Encoding encoding)
+        ValueTask<string> ReadNullTerminatedString(Encoding encoding, bool async, CancellationToken cancellationToken = default)
         {
-            int i;
-            for (i = ReadPosition; Buffer[i] != 0; i++)
-                Debug.Assert(i <= ReadPosition + ReadBytesLeft);
-            Debug.Assert(i >= ReadPosition);
-            var result = encoding.GetString(Buffer, ReadPosition, i - ReadPosition);
-            ReadPosition = i + 1;
-            return result;
+            return ReadFromBuffer(out var s)
+                ? new ValueTask<string>(s)
+                : ReadLong(s);
+
+            bool ReadFromBuffer(out string s)
+            {
+                var start = ReadPosition;
+                while (ReadPosition < FilledBytes)
+                {
+                    if (Buffer[ReadPosition++] == 0)
+                    {
+                        s = encoding.GetString(Buffer, start, ReadPosition - start - 1);
+                        return true;
+                    }
+                }
+
+                s = encoding.GetString(Buffer, start, ReadPosition - start);
+                return false;
+            }
+
+            async ValueTask<string> ReadLong(string s)
+            {
+                var builder = new StringBuilder(s);
+                bool complete;
+                do
+                {
+                    await ReadMore(async, cancellationToken);
+                    complete = ReadFromBuffer(out s);
+                    builder.Append(s);
+                }
+                while (!complete);
+
+                return builder.ToString();
+            }
         }
 
         public ReadOnlySpan<byte> GetNullTerminatedBytes()

src/Npgsql/NpgsqlTypes/NpgsqlDbType.cs

@@ -484,6 +484,17 @@ public enum NpgsqlDbType
         [BuiltInPostgresType("tid", PostgresTypeOIDs.Tid)]
         Tid = 53,
 
+        /// <summary>
+        /// Corresponds to the PostgreSQL "pg_lsn" type, which can be used to store LSN (Log Sequence Number) data which
+        /// is a pointer to a location in the WAL.
+        /// </summary>
+        /// <remarks>
+        /// See: https://www.postgresql.org/docs/current/datatype-pg-lsn.html and
+        /// https://git.postgresql.org/gitweb/?p=postgresql.git;a=commit;h=7d03a83f4d0736ba869fa6f93973f7623a27038a
+        /// </remarks>
+        [BuiltInPostgresType("pg_lsn", 3220)]
+        PgLsn = 59,
+
         #endregion
 
         #region Special