Add ParameterStatus tracking

.release-notes/parameter-status.md

@@ -0,0 +1,13 @@
+## Add ParameterStatus tracking
+
+PostgreSQL sends ParameterStatus messages during connection startup to report runtime parameter values (server_version, client_encoding, standard_conforming_strings, etc.) and again whenever a `SET` command changes a reporting parameter. Previously, the driver silently discarded these messages.
+
+A new `pg_parameter_status` callback on `SessionStatusNotify` delivers each parameter as a `ParameterStatus` value with `name` and `value` fields:
+
+```pony
+actor MyNotify is SessionStatusNotify
+  be pg_parameter_status(session: Session, status: ParameterStatus) =>
+    _env.out.print(status.name + " = " + status.value)
+```
+
+The callback has a default no-op implementation, so existing code is unaffected.

examples/parameter-status/parameter-status-example.pony

@@ -0,0 +1,77 @@
+use "cli"
+use "collections"
+use lori = "lori"
+// in your code this `use` statement would be:
+// use "postgres"
+use "../../postgres"
+
+actor Main
+  new create(env: Env) =>
+    let server_info = ServerInfo(env.vars)
+    let auth = lori.TCPConnectAuth(env.root)
+
+    let client = Client(auth, server_info, env.out)
+
+// This example demonstrates ParameterStatus tracking. It prints server
+// parameters as they arrive during startup, then executes a SET command
+// to change application_name and prints the updated value.
+actor Client is (SessionStatusNotify & ResultReceiver)
+  let _session: Session
+  let _out: OutStream
+
+  new create(auth: lori.TCPConnectAuth, info: ServerInfo, out: OutStream) =>
+    _out = out
+    _session = Session(
+      ServerConnectInfo(auth, info.host, info.port),
+      DatabaseConnectInfo(info.username, info.password, info.database),
+      this)
+
+  be close() =>
+    _session.close()
+
+  be pg_session_authenticated(session: Session) =>
+    _out.print("Authenticated.")
+    _out.print("Setting application_name...")
+    session.execute(
+      SimpleQuery("SET application_name = 'pony_example'"), this)
+
+  be pg_session_authentication_failed(
+    s: Session,
+    reason: AuthenticationFailureReason)
+  =>
+    _out.print("Failed to authenticate.")
+
+  be pg_parameter_status(session: Session, status: ParameterStatus) =>
+    _out.print("Parameter: " + status.name + " = " + status.value)
+
+  be pg_query_result(session: Session, result: Result) =>
+    _out.print("SET completed.")
+    _out.print("Done.")
+    close()
+
+  be pg_query_failed(session: Session, query: Query,
+    failure: (ErrorResponseMessage | ClientQueryError))
+  =>
+    match failure
+    | let e: ErrorResponseMessage =>
+      _out.print("Query failed: [" + e.severity + "] " + e.code + ": "
+        + e.message)
+    | let e: ClientQueryError =>
+      _out.print("Query failed: client error")
+    end
+    close()
+
+class val ServerInfo
+  let host: String
+  let port: String
+  let username: String
+  let password: String
+  let database: String
+
+  new val create(vars: (Array[String] val | None)) =>
+    let e = EnvVars(vars)
+    host = try e("POSTGRES_HOST")? else "127.0.0.1" end
+    port = try e("POSTGRES_PORT")? else "5432" end
+    username = try e("POSTGRES_USERNAME")? else "postgres" end
+    password = try e("POSTGRES_PASSWORD")? else "postgres" end
+    database = try e("POSTGRES_DATABASE")? else "postgres" end

postgres/_backend_messages.pony

@@ -154,6 +154,19 @@ class val _ParameterDescriptionMessage
   new val create(param_oids': Array[U32] val) =>
     param_oids = param_oids'
 
+class val _ParameterStatusMessage
+  """
+  Message from the backend reporting a runtime parameter name and its current
+  value. Sent during startup for all reporting parameters, and again whenever
+  a SET command changes one.
+  """
+  let name: String
+  let value: String
+
+  new val create(name': String, value': String) =>
+    name = name'
+    value = value'
+
 class val _NotificationResponseMessage
   """
   Message from the backend delivering a LISTEN/NOTIFY notification.

postgres/_response_message_parser.pony

@@ -59,9 +59,8 @@ primitive _ResponseMessageParser
           s.state.on_notice(s, msg)
         | let msg: _CopyInResponseMessage =>
           s.state.on_copy_in_response(s, msg)
-        | _SkippedMessage =>
-          // Known async message (ParameterStatus) — intentionally not routed.
-          None
+        | let msg: _ParameterStatusMessage =>
+          s.state.on_parameter_status(s, msg)
         | let msg: _EmptyQueryResponseMessage =>
           s.state.on_empty_query_response(s)
         | None =>

postgres/_response_parser.pony

@@ -25,28 +25,17 @@ type _ResponseParserResult is
   | _NotificationResponseMessage
   | _CopyInResponseMessage
   | _PortalSuspendedMessage
-  | _SkippedMessage
+  | _ParameterStatusMessage
   | _UnsupportedMessage
   | ErrorResponseMessage
   | NoticeResponseMessage
   | None )
 
-primitive _SkippedMessage
-  """
-  Returned by the parser for known PostgreSQL asynchronous message types that
-  the driver recognizes but intentionally does not process: ParameterStatus.
-  These can arrive between any other messages and are safely ignored.
-
-  Distinct from `_UnsupportedMessage`, which represents truly unknown message
-  types that the parser does not recognize at all.
-  """
-
 primitive _UnsupportedMessage
   """
-  Returned by the parser for message types that are not recognized. This
-  represents truly unknown messages — not messages the driver intentionally
-  skips (those return `_SkippedMessage`). A future PostgreSQL version could
-  introduce new message types that would hit this path.
+  Returned by the parser for message types that are not recognized. A future
+  PostgreSQL version could introduce new message types that would hit this
+  path.
   """
 
 primitive _ResponseParser
@@ -206,9 +195,11 @@ primitive _ResponseParser
       let secret_key = buffer.i32_be()?
       return _BackendKeyDataMessage(process_id, secret_key)
     | _MessageType.parameter_status() =>
-      // Known async message — skip payload without parsing
-      buffer.skip(message_size)?
-      return _SkippedMessage
+      // Slide past the header...
+      buffer.skip(5)?
+      // and parse the parameter status payload in an isolated reader
+      let ps_payload = buffer.block(payload_size)?
+      return _parameter_status(consume ps_payload)?
     | _MessageType.notice_response() =>
       // Slide past the header...
       buffer.skip(5)?
@@ -391,6 +382,20 @@ primitive _ResponseParser
       _CommandCompleteMessage(id, 0)
     end
 
+  fun _parameter_status(payload: Array[U8] val)
+    : _ParameterStatusMessage ?
+  =>
+    """
+    Parse a ParameterStatus message. Payload is two null-terminated strings:
+    parameter name followed by parameter value.
+    """
+    let reader: Reader = Reader.>append(payload)
+    let name_bytes = reader.read_until(0)?
+    let name = String.from_array(consume name_bytes)
+    let value_bytes = reader.read_until(0)?
+    let value = String.from_array(consume value_bytes)
+    _ParameterStatusMessage(name, value)
+
   fun _notification_response(payload: Array[U8] val)
     : _NotificationResponseMessage ?
   =>

postgres/_test.pony

@@ -52,7 +52,7 @@ actor \nodoc\ Main is TestList
     test(_TestResponseParserDigitMessageTypeNotJunk)
     test(_TestResponseParserMultipleMessagesParseCompleteFirst)
     test(_TestResponseParserMultipleMessagesBackendKeyDataFirst)
-    test(_TestResponseParserParameterStatusSkipped)
+    test(_TestResponseParserParameterStatusMessage)
     test(_TestResponseParserNoticeResponseMessage)
     test(_TestResponseParserNotificationResponseMessage)
     test(_TestResponseParserMultipleMessagesAsyncThenAuth)
@@ -143,6 +143,10 @@ actor \nodoc\ Main is TestList
     test(_TestNoticeDelivery)
     test(_TestNoticeDuringDataRows)
     test(_TestNoticeOnDropIfExists)
+    test(_TestParameterStatusDelivery)
+    test(_TestParameterStatusDuringDataRows)
+    test(_TestParameterStatusOnStartup)
+    test(_TestParameterStatusOnSet)
     test(_TestResponseParserCopyInResponseMessage)
     test(_TestFrontendMessageCopyData)
     test(_TestFrontendMessageCopyDone)