Pin the time representation of peer sessions (SPOC-660)

src/spock.c

@@ -194,6 +194,8 @@ static PGconn *spock_connect_base(const char *connstr,
 								  const char *appname,
 								  const char *suffix,
 								  bool replication);
+static char *spock_get_conninfo_option(const char *connstr,
+									   const char *keyword);
 
 /*
  * Ensure string is not longer than maxlen.
@@ -351,7 +353,12 @@ get_spock_table_oid(const char *table)
 	return reloid;
 }
 
-#define CONN_PARAM_ARRAY_SIZE 10
+/*
+ * One more than the number of parameters we pass, to leave room for the NULL
+ * terminator.  In passing, bumped from 10 when "options" was added; the array
+ * was previously an exact fit, which left no margin for the next parameter.
+ */
+#define CONN_PARAM_ARRAY_SIZE 11
 
 static PGconn *
 spock_connect_base(const char *connstr, const char *appname,
@@ -362,6 +369,8 @@ spock_connect_base(const char *connstr, const char *appname,
 	const char *keys[CONN_PARAM_ARRAY_SIZE];
 	const char *vals[CONN_PARAM_ARRAY_SIZE];
 	char		appname_buf[NAMEDATALEN];
+	char	   *user_options;
+	char	   *options_val;
 	StringInfoData s;
 
 	initStringInfo(&s);
@@ -402,16 +411,44 @@ spock_connect_base(const char *connstr, const char *appname,
 	keys[i] = "replication";
 	vals[i] = replication ? "database" : NULL;
 	i++;
+
+	/*
+	 * Force assorted GUC parameters to settings that ensure that the peer will
+	 * output data values in a form that is unambiguous to us when the transfer
+	 * happens in text mode.  (We don't touch our own settings; that would
+	 * surprise user-defined code running here, such as triggers.)  ISO is the
+	 * only DateStyle that always emits an unambiguous date and an explicit
+	 * numeric UTC offset, which is also why TimeZone needs no forcing.
+	 *
+	 * Subscription sync reads spock.progress from a peer over a regular
+	 * connection, so this is applied to those as well, not only to replication
+	 * ones.
+	 *
+	 * Caution: an explicit array entry overrides any "options" carried by the
+	 * expanded dbname string, so the user's must be prepended rather than
+	 * replaced.  Ours win by coming last, as the backend applies -c in order.
+	 */
+	user_options = spock_get_conninfo_option(connstr, "options");
+	options_val = psprintf("%s -c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3",
+						   (user_options == NULL) ? "" : user_options);
+	if (user_options != NULL)
+		pfree(user_options);
+
+	keys[i] = "options";
+	vals[i] = options_val;
+	i++;
+
 	keys[i] = NULL;
 	vals[i] = NULL;
 
-	Assert(i <= CONN_PARAM_ARRAY_SIZE);
+	Assert(i < CONN_PARAM_ARRAY_SIZE);
 
 	/*
 	 * We use the expand_dbname parameter to process the connection string (or
 	 * URI), and pass some extra options.
 	 */
 	conn = PQconnectdbParams(keys, vals, /* expand_dbname = */ true);
+
 	if (PQstatus(conn) != CONNECTION_OK)
 	{
 		ereport(ERROR,
@@ -421,11 +458,73 @@ spock_connect_base(const char *connstr, const char *appname,
 				 errdetail("dsn was: %s", s.data)));
 	}
 
+	/*
+	 * Released only once nothing can read vals[] again.  libpq copied what it
+	 * needed during the call above, so freeing earlier would work, but it would
+	 * leave a dangling entry in vals[] across the error path -- and that path
+	 * already formats connection details for the report.
+	 */
+	pfree(options_val);
+
 	resetStringInfo(&s);
 
 	return conn;
 }
 
+/*
+ * spock_get_conninfo_option
+ *		Return the value of a single option from a connection string, or NULL
+ *		if it is absent or empty.
+ *
+ *		Result is palloc'd in CurrentMemoryContext.  If the option appears more
+ *		than once the last occurrence wins, matching libpq's own precedence.
+ *
+ *		Modelled on libpqrcv_get_option_from_conninfo() in
+ *		src/backend/replication/libpqwalreceiver/libpqwalreceiver.c, but
+ *		deliberately does not raise an error on an unparseable string; see below.
+ */
+static char *
+spock_get_conninfo_option(const char *connstr, const char *keyword)
+{
+	PQconninfoOption *opts;
+	PQconninfoOption *opt;
+	char	   *option = NULL;
+	char	   *err = NULL;
+
+	Assert(connstr != NULL);
+	Assert(keyword != NULL);
+
+	opts = PQconninfoParse(connstr, &err);
+	if (opts == NULL)
+	{
+		/*
+		 * Deliberately not an error.  PQconnectdbParams() takes a dbname with
+		 * neither '=' nor a URI prefix literally, so a DSN of plain "mydb"
+		 * connects today while PQconninfoParse() rejects it; raising an error
+		 * here would break such configurations.  Report no options instead, and
+		 * let PQconnectdbParams() complain if the string really is malformed.
+		 *
+		 * The error string is malloc'd, so we must free it explicitly.
+		 */
+		PQfreemem(err);
+		return NULL;
+	}
+
+	for (opt = opts; opt->keyword != NULL; opt++)
+	{
+		if (strcmp(opt->keyword, keyword) == 0 && opt->val && opt->val[0] != '\0')
+		{
+			if (option != NULL)
+				pfree(option);
+			option = pstrdup(opt->val);
+		}
+	}
+
+	PQconninfoFree(opts);
+
+	return option;
+}
+
 
 /*
  * Make standard postgres connection, ERROR on failure.

src/spock_common.c

@@ -21,6 +21,7 @@
 #include "storage/lmgr.h"
 #include "storage/proc.h"
 #include "utils/builtins.h"
+#include "utils/datetime.h"
 #include "utils/guc.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
@@ -668,13 +669,128 @@ write_buf(int fd, const void *buf, size_t nbytes, const char *filename)
 	}
 }
 
+/*
+ * commit_ts_str_has_offset
+ *		Does this commit timestamp text representation carry an explicit UTC
+ *		offset?
+ *
+ *		Uses ParseDateTime(), the same tokeniser timestamptz_in() runs, so that
+ *		this predicate cannot drift away from what the parser actually accepts.
+ *		Hand-rolled scanning was tried first and got the era suffix wrong: ISO
+ *		output for a BC date is "0044-03-15 11:45:16-00:14:44 BC", where the
+ *		offset is not the last field.
+ *
+ *		Only the presence of an offset is decided here.  Whether the value
+ *		denotes a real instant is a separate question, and the caller answers it
+ *		separately; do not fold the two together.
+ */
+static bool
+commit_ts_str_has_offset(const char *s)
+{
+	char		workbuf[MAXDATELEN + MAXDATEFIELDS];
+	char	   *field[MAXDATEFIELDS];
+	int			ftype[MAXDATEFIELDS];
+	int			nf;
+	int			i;
+
+	Assert(s != NULL);
+
+	/*
+	 * Not tokenisable as a datetime at all.  The caller parses before asking, so
+	 * a string that reaches us has already satisfied timestamptz_in() and this
+	 * should not happen; answer no rather than claim an offset we did not see.
+	 * Note the contract is that negative means failure, zero or positive
+	 * success.
+	 */
+	if (ParseDateTime(s, workbuf, sizeof(workbuf), field, ftype,
+					  MAXDATEFIELDS, &nf) < 0)
+		return false;
+
+	/*
+	 * DTK_TZ is a numeric UTC offset.  A zone abbreviation arrives as
+	 * DTK_STRING instead and is deliberately not accepted here: an abbreviation
+	 * is not unique across zones -- CST is China, Cuba and US Central -- so it
+	 * does not pin down an instant even when the reader can resolve it.
+	 */
+	for (i = 0; i < nf; i++)
+	{
+		if (ftype[i] == DTK_TZ)
+			return true;
+	}
+
+	return false;
+}
+
+/*
+ * str_to_timestamptz
+ *		Parse a commit timestamp that reached us from another node.
+ *
+ *		Deliberately not a general timestamptz parser.  Its only inputs are
+ *		spock.progress.remote_commit_ts and .last_updated_ts, read from a peer
+ *		over libpq, and both are instants stamped by a running server.  A finite
+ *		value bearing an explicit offset is the only thing that can legitimately
+ *		arrive, so everything else is refused rather than interpreted.
+ *
+ *		Peer sessions are started with datestyle=ISO (see spock_connect_base()),
+ *		so the offset should always be there.  If it is not, something between us
+ *		and the peer has changed the representation; timestamptz_in() would then
+ *		quietly resolve the value in the local TimeZone and move the instant, so
+ *		refuse rather than guess.
+ *
+ *		Both checks should be unreachable in a correctly configured cluster.
+ *		They are ereport() and not Assert() on purpose: this string arrives over
+ *		the network, which makes it input to be validated rather than an internal
+ *		invariant, and an Assert would be compiled out of exactly the production
+ *		builds where a mangled value does its damage.
+ *
+ *		Note the order.  Parsing comes first so that a malformed string draws
+ *		timestamptz_in()'s own complaint, which is better than anything we could
+ *		phrase, and so that "infinity" is diagnosed as the non-instant it is
+ *		rather than as a missing time zone.  Checking the offset first looked
+ *		tidier but made the finiteness test unreachable: no spelling of infinity
+ *		carries an offset, so it was always rejected one branch earlier, under a
+ *		message that would have sent the reader hunting through the peer's
+ *		DateStyle for nothing.  Parsing a zone-less value here is harmless; the
+ *		result is discarded before anyone can use it.
+ */
 TimestampTz
 str_to_timestamptz(const char *s)
 {
-	return DatumGetTimestampTz(DirectFunctionCall3(timestamptz_in,
-												   CStringGetDatum(s),
-												   ObjectIdGetDatum(InvalidOid),
-												   Int32GetDatum(-1)));
+	TimestampTz result;
+
+	result = DatumGetTimestampTz(DirectFunctionCall3(timestamptz_in,
+													CStringGetDatum(s),
+													ObjectIdGetDatum(InvalidOid),
+													Int32GetDatum(-1)));
+
+	/*
+	 * Rejects infinity and -infinity, which no server stamps a commit with.
+	 * Letting one through would put INT64_MAX or INT64_MIN into
+	 * SpockApplyProgress and from there into spock.progress, where every lag
+	 * figure and every prev_remote_ts comparison derived from it turns into
+	 * nonsense.  spock_sync.c already Asserts IS_VALID_TIMESTAMP on the result;
+	 * this makes the same check hold in production builds, at the one place all
+	 * such values pass through.
+	 */
+	if (!IS_VALID_TIMESTAMP(result))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_DATETIME_FORMAT),
+				 errmsg("commit timestamp \"%s\" received from a peer node is not a finite instant",
+						s)));
+
+	if (!commit_ts_str_has_offset(s))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_DATETIME_FORMAT),
+				 errmsg("commit timestamp \"%s\" received from a peer node carries no time zone",
+						s),
+				 errdetail("A commit timestamp denotes an absolute instant; "
+						   "interpreting this one in the local time zone would "
+						   "shift it."),
+				 errhint("The peer session is expected to run with "
+						 "datestyle=ISO, which Spock requests when it "
+						 "connects.")));
+
+	return result;
 }
 
 XLogRecPtr

tests/tap/schedule

@@ -54,6 +54,7 @@ test: 030_autoddl_repset_stickiness
 test: 032_lolor_largeobject_repset
 test: 033_zodan_lolor_add_node
 test: 034_reserved_object_ddl
+test: 037_wire_format_datestyle
 test: 044_apply_change_logging
 test: 045_lsn_from_commit_ts
 # Upgrade schema match test (builds from source, slow):