makes clearer pg timestamps should not be used for ordering txs (#2832)

Author: vroldanbet

internal/datastore/postgres/postgres.go

@@ -503,7 +503,7 @@ func (pgd *pgDatastore) ReadWriteTx(
 			return nil, spiceerrors.MustBugf("could not cast timestamp to uint64")
 		}
 
-		return postgresRevision{snapshot: newSnapshot.markComplete(newXID.Uint64), optionalTxID: newXID, optionalNanosTimestamp: nanosTimestamp}, nil
+		return postgresRevision{snapshot: newSnapshot.markComplete(newXID.Uint64), optionalTxID: newXID, optionalInexactNanosTimestamp: nanosTimestamp}, nil
 	}
 
 	if !config.DisableRetries {

internal/datastore/postgres/postgres_shared_test.go

@@ -455,7 +455,7 @@ func ReadWriteTxReturnsOptionalRevisionFields(t *testing.T, ds datastore.Datasto
 	require.NoError(err)
 	pgRev, ok := rev.(postgresRevision)
 	require.True(ok)
-	require.Positive(pgRev.optionalNanosTimestamp, "revision timestamp should be set")
+	require.Positive(pgRev.optionalInexactNanosTimestamp, "revision timestamp should be set")
 	require.True(pgRev.optionalTxID.Valid, "revision txid be set")
 }
 
@@ -977,8 +977,8 @@ func OverlappingRevisionTest(t *testing.T, b testdatastore.RunningEngineForTest)
 			5 * time.Second,
 			0,
 			[]postgresRevision{
-				{optionalTxID: NewXid8(3), snapshot: pgSnapshot{xmin: 1, xmax: 4, xipList: []uint64{2}}, optionalNanosTimestamp: uint64((time.Second * 1) * time.Nanosecond)},
-				{optionalTxID: NewXid8(2), snapshot: pgSnapshot{xmin: 1, xmax: 4, xipList: []uint64{3}}, optionalNanosTimestamp: uint64((time.Second * 2) * time.Nanosecond)},
+				{optionalTxID: NewXid8(3), snapshot: pgSnapshot{xmin: 1, xmax: 4, xipList: []uint64{2}}, optionalInexactNanosTimestamp: uint64((time.Second * 1) * time.Nanosecond)},
+				{optionalTxID: NewXid8(2), snapshot: pgSnapshot{xmin: 1, xmax: 4, xipList: []uint64{3}}, optionalInexactNanosTimestamp: uint64((time.Second * 2) * time.Nanosecond)},
 			},
 			2, 0,
 		},
@@ -1022,7 +1022,7 @@ func OverlappingRevisionTest(t *testing.T, b testdatastore.RunningEngineForTest)
 				stmt := sq.StatementBuilder.PlaceholderFormat(sq.Dollar)
 				insertTxn := stmt.Insert(schema.TableTransaction).Columns(schema.ColXID, schema.ColSnapshot, schema.ColTimestamp)
 
-				ts := time.Unix(0, int64(rev.optionalNanosTimestamp)) //nolint:gosec
+				ts := time.Unix(0, int64(rev.optionalInexactNanosTimestamp)) //nolint:gosec
 				sql, args, err := insertTxn.Values(rev.optionalTxID, rev.snapshot, ts).ToSql()
 				require.NoError(err)
 

internal/datastore/postgres/revisions.go

@@ -229,8 +229,8 @@ func parseRevisionProto(revisionStr string) (datastore.Revision, error) {
 			xmax:    xmax,
 			xipList: xips,
 		},
-		optionalTxID:           xid8{Uint64: decoded.OptionalTxid, Valid: decoded.OptionalTxid != 0},
-		optionalNanosTimestamp: decoded.OptionalTimestamp,
+		optionalTxID:                  xid8{Uint64: decoded.OptionalTxid, Valid: decoded.OptionalTxid != 0},
+		optionalInexactNanosTimestamp: decoded.OptionalTimestamp,
 	}, nil
 }
 
@@ -318,10 +318,14 @@ func createNewTransaction(ctx context.Context, tx pgx.Tx, metadata map[string]an
 }
 
 type postgresRevision struct {
-	snapshot               pgSnapshot
-	optionalTxID           xid8
-	optionalNanosTimestamp uint64
-	optionalMetadata       dscommon.TransactionMetadata
+	snapshot     pgSnapshot
+	optionalTxID xid8
+	// timestamps cannot be used for ordering transactions in PG, but are useful to get a rough estimate of
+	// when did the transaction happen (e.g. for lag-computation purposes on Watch API consumers).
+	// For ordering purposes, only the snapshot values should be used (and it's still possible for two snapshots to
+	// be overlapping)
+	optionalInexactNanosTimestamp uint64
+	optionalMetadata              dscommon.TransactionMetadata
 }
 
 func (pr postgresRevision) ByteSortable() bool {
@@ -374,13 +378,14 @@ func (pr postgresRevision) OptionalTransactionID() (xid8, bool) {
 }
 
 // OptionalNanosTimestamp returns a unix epoch timestamp in nanos representing the time at which the transaction committed
-// as defined by the Postgres primary. This is not guaranteed to be monotonically increasing
+// as defined by the Postgres primary. This is not guaranteed to be monotonically increasing, and thus should
+// not be used for ordering transactions.
 func (pr postgresRevision) OptionalNanosTimestamp() (uint64, bool) {
-	if pr.optionalNanosTimestamp == 0 {
+	if pr.optionalInexactNanosTimestamp == 0 {
 		return 0, false
 	}
 
-	return pr.optionalNanosTimestamp, true
+	return pr.optionalInexactNanosTimestamp, true
 }
 
 // MarshalBinary creates a version of the snapshot that uses relative encoding
@@ -409,7 +414,7 @@ func (pr postgresRevision) MarshalBinary() ([]byte, error) {
 		RelativeXmax:      relativeXmax - xminInt,
 		RelativeXips:      relativeXips,
 		OptionalTxid:      pr.optionalTxID.Uint64,
-		OptionalTimestamp: pr.optionalNanosTimestamp,
+		OptionalTimestamp: pr.optionalInexactNanosTimestamp,
 	}
 
 	return protoRevision.MarshalVT()

internal/datastore/postgres/revisions_test.go

@@ -81,9 +81,9 @@ func TestRevisionSerDe(t *testing.T) {
 			require := require.New(t)
 
 			rev := postgresRevision{
-				snapshot:               tc.snapshot,
-				optionalTxID:           xid8{Uint64: tc.optionalTxID, Valid: tc.optionalTxID > 0},
-				optionalNanosTimestamp: tc.optionalNanoTS,
+				snapshot:                      tc.snapshot,
+				optionalTxID:                  xid8{Uint64: tc.optionalTxID, Valid: tc.optionalTxID > 0},
+				optionalInexactNanosTimestamp: tc.optionalNanoTS,
 			}
 			serialized := rev.String()
 			require.Equal(tc.expectedStr, serialized)
@@ -99,7 +99,7 @@ func TestTxIDTimestampAvailable(t *testing.T) {
 	// Timestamps should be non-negative
 	testTimestamp := safecast.RequireConvert[uint64](t, time.Now().Unix())
 	snapshot := snap(0, 5, 1)
-	pgr := postgresRevision{snapshot: snapshot, optionalTxID: NewXid8(1), optionalNanosTimestamp: testTimestamp}
+	pgr := postgresRevision{snapshot: snapshot, optionalTxID: NewXid8(1), optionalInexactNanosTimestamp: testTimestamp}
 	receivedTimestamp, ok := pgr.OptionalNanosTimestamp()
 	require.True(t, ok)
 	require.Equal(t, receivedTimestamp, testTimestamp)

internal/datastore/postgres/watch.go

@@ -173,9 +173,9 @@ func (pgd *pgDatastore) Watch(
 				currentTxn = newTxns[len(newTxns)-1]
 				for _, newTx := range newTxns {
 					currentTxn = postgresRevision{
-						snapshot:               currentTxn.snapshot.markComplete(newTx.optionalTxID.Uint64),
-						optionalTxID:           currentTxn.optionalTxID,
-						optionalNanosTimestamp: currentTxn.optionalNanosTimestamp,
+						snapshot:                      currentTxn.snapshot.markComplete(newTx.optionalTxID.Uint64),
+						optionalTxID:                  currentTxn.optionalTxID,
+						optionalInexactNanosTimestamp: currentTxn.optionalInexactNanosTimestamp,
 					}
 				}
 
@@ -229,10 +229,10 @@ func (pgd *pgDatastore) getNewRevisions(ctx context.Context, afterTX postgresRev
 			}
 
 			ids = append(ids, postgresRevision{
-				snapshot:               nextSnapshot.markComplete(nextXID.Uint64),
-				optionalTxID:           nextXID,
-				optionalNanosTimestamp: nanosTimestamp,
-				optionalMetadata:       metadata,
+				snapshot:                      nextSnapshot.markComplete(nextXID.Uint64),
+				optionalTxID:                  nextXID,
+				optionalInexactNanosTimestamp: nanosTimestamp,
+				optionalMetadata:              metadata,
 			})
 		}
 		if rows.Err() != nil {