Replication, Partitioning, and Sharding Questions

Direct answer

A single-shard failure in a sharded SQL database only affects the subset of users whose data lives on that shard, which is both the good news (the blast radius is bounded) and the trap (it is easy to under-react to an outage that "only" affects a fraction of users but is, for those users, a full outage). The playbook is: detect which shard and confirm the primary is actually unreachable (not just slow), fence it so it cannot accept writes if it comes back unexpectedly, promote the most caught-up in-sync replica, repoint the routing layer's shard map, then run data-integrity checks before declaring the incident over.

The playbook

flowchart TD
  A[Alert: elevated errors on shard 7] --> B{Is shard 7 primary reachable?}
  B -->|no| C[Fence old primary: revoke credentials, block writes]
  C --> D[Promote most-caught-up in-sync replica to primary for shard 7]
  D --> E[Repoint routing layer: shard map entry for 7 -> new primary]
  B -->|yes, but degraded| F[Diagnose: disk, connections, replication lag on shard 7]
  F --> G{Root cause fixable in place?}
  G -->|yes| H[Mitigate in place, no promotion]
  G -->|no| C
  E --> I[Run data-integrity checks: row counts, checksums vs other shards' expectations]
  H --> I
  I --> J[Validate application-level reads/writes on shard 7 succeed]
  J --> K[Post-recovery: reconcile any writes lost in the gap, document timeline]

1. Detection. The alert should identify the specific shard, not just "database errors," which means shard-aware monitoring (per-shard error rate, per-shard connection health, per-shard replication lag, meaning how far behind the primary this shard's replica has fallen) has to exist before the incident, not be built during it. Confirm scope immediately: query the shard-to-tenant or shard-to-user-range mapping to know exactly which users are affected, since this drives both the urgency (how many users, and are any high-priority accounts among them) and the communication (what to tell support or status pages).

2. Immediate mitigation. First check whether the primary is truly unreachable versus merely degraded (high load, a stuck query, disk pressure); a degraded-but-reachable primary may be fixable in place (kill a runaway query, add I/O capacity) without the risk of a promotion. If it is genuinely unreachable, fence it (revoke its credentials or otherwise ensure it cannot accept writes if it unexpectedly comes back mid-recovery, which prevents a split-brain where two nodes for the same shard both think they are primary), then promote the most caught-up in-sync replica for that shard specifically, and update the routing layer's shard map so application traffic for that shard's key range goes to the new primary.

3. Data-integrity checks. Before declaring the shard healthy, verify: row counts and, where feasible, checksums on recently-written tables against expectations (comparing against a recent backup or against another shard's equivalent tables if the schema allows structural comparison); confirm the promoted replica's applied position (its last-replayed log sequence number) to bound how many, if any, in-flight writes from the old primary may not have made it across, which tells you the actual recovery point objective (RPO) hit for this specific incident, not a theoretical one; and run a small set of representative application-level read/write operations end-to-end against the shard, not just database-level health checks, since a shard can be database-healthy while still broken from the application's perspective (wrong connection string, stale cached routing entry).

4. Post-recovery validation. Reconcile any writes that were in flight but unconfirmed at the moment of failure (check application-level idempotency logs or an outbox table if one exists, to identify anything that needs replaying or compensating). Document the actual timeline (detection time, fencing time, promotion time, full recovery time) against the shard's specific RTO target, and capture the RPO actually incurred (how many seconds, or how many specific rows, of writes were lost, if any) for the post-mortem, since "we recovered" and "we recovered with zero data loss" are different claims that need different evidence.

Trade-offs and pitfalls

The most common mistake is skipping the fencing step under time pressure ("it's clearly dead, let's just promote"), which is exactly the scenario that produces split-brain if the old primary was actually just network-partitioned rather than dead, and comes back mid-incident still believing it is primary. The second common mistake is declaring victory once the promoted replica is serving traffic without running the data-integrity checks, which defers discovering data loss or corruption from "during the incident, with full attention on it" to "days later, when a user reports a missing record," at which point root-causing it is far harder.

Direct answer

Split brain is best treated as a prevention problem first and a cleanup problem second: quorum enforcement and fencing tokens are what stop two nodes from both believing they are the writer in the first place, and they are cheap and mechanical compared to reconciling conflicting writes after the fact, which always either loses data or needs application-specific merge logic. Detect the split from the symptom that actually matters (a minority-side node continuing to accept writes without quorum), fence the loser the moment a new leader is elected, and only fall back to conflict resolution (last-write-wins or a merge function) for the writes that already slipped through before the fence took effect.

Detecting the split

A network partition means each side can normally reach its own local replicas but not the other side's. The mechanism that catches this reliably is quorum: every write and (in a strict system) every read must be acknowledged by a majority of the full replica set, not just "whatever this node can currently see." A minority-side node that keeps accepting writes without checking for a majority is the actual failure, whether or not anyone notices immediately. Detection signals worth alerting on: a node's write quorum acknowledgment rate dropping below the majority threshold, two nodes both reporting themselves as the current leader (directly observable if leadership is published through the same coordination service, like a lease key), and a spike in application-level conflict errors once the partition heals and replicas start reconciling.

Preventive controls (the part that matters most)

Quorum enforcement. Require a majority of replicas to acknowledge a write before it is considered committed, and require the same majority for the leader election itself. With quorum-based leadership, at most one side of any partition can hold a majority, so at most one side can ever elect a leader and accept writes. The minority side simply cannot make progress until the partition heals, which is a feature: it is unavailable, not wrong.

Fencing tokens. Quorum protects new writes, but it does not stop a leader that already believed it was in charge from finishing a write it started before losing quorum (a leader that stalled on a long garbage-collection pause, for example, has no way to know it lost its lease). A fencing token is a monotonically increasing number issued by the coordination service every time leadership changes hands. Every write carries the current token, and storage rejects any write whose token is lower than the highest token it has already seen. This closes the exact gap that a network partition or a slow node opens up: a stale leader's write is rejected even though the leader itself does not know it is stale.

After it happens anyway: reconciliation and conflict resolution policy

For the small window of writes that land before the fence engages (or in a system that runs multi-primary by design rather than single-leader), you need an explicit conflict resolution policy, and there is a real trade-off between the two common choices:

• Last-write-wins (by timestamp). Simple, requires no application logic, and is what most managed multi-region databases default to. The cost: it silently discards one of the two writes, with no record that a conflict occurred unless you build that logging yourself, and it is wrong for any field where "both changes should count" (an inventory decrement, a counter, a set of items added).
• Application-specific merge. Correct for the cases last-write-wins gets wrong (merge two carts instead of picking one, union two tag sets, keep the higher of two account balances after a dispute), but it means writing and testing bespoke merge code for every mutable field that can conflict, and someone has to maintain it as the schema evolves.

Recommendation: default to last-write-wins for fields where losing a write is a minor inconvenience (a display name, a last-seen timestamp), and require an explicit merge function for anything where losing a write is a real incident (money, inventory counts, anything with financial or contractual consequences). Never apply last-write-wins blindly across an entire record; decide per field.

Worked example: what a fencing token actually stops

class FencedStorage:
    """A storage node that enforces fencing tokens: it only accepts a write if the
    token attached to it is >= the highest token it has ever seen. This is the whole
    mechanism; the storage node does not need to know anything about leader election."""
    def __init__(self):
        self.highest_token_seen = -1
        self.value = None
        self.rejected_writes = []

    def write(self, token: int, value: str, writer_label: str):
        if token < self.highest_token_seen:
            self.rejected_writes.append((writer_label, token, value))
            return False
        self.highest_token_seen = token
        self.value = value
        return True


# a lock/lease service hands out monotonically increasing tokens on each acquisition
storage = FencedStorage()

# leader A acquires the lease, gets token 33, then stalls (long GC pause / network
# partition) BEFORE its write reaches storage
token_a = 33

# the lease expires while A is stalled; leader B acquires the lease next, gets the
# next token, and successfully writes
token_b = 34
ok_b = storage.write(token_b, "value-from-B", "leader-B")

# A wakes up, has no idea it lost the lease, and sends its stale write with its OLD
# token
ok_a = storage.write(token_a, "value-from-A (stale)", "leader-A")

print(f"leader B (token={token_b}) write accepted: {ok_b}, storage value now: {storage.value!r}")
print(f"leader A (token={token_a}) write accepted: {ok_a}")
print(f"storage's highest_token_seen: {storage.highest_token_seen}")
print(f"final stored value: {storage.value!r}")
print(f"rejected writes: {storage.rejected_writes}")
print()
print("-- for contrast, the SAME scenario WITHOUT fencing (storage accepts any write, "
      "no token check) --")
class UnfencedStorage:
    def __init__(self):
        self.value = None

    def write(self, value: str):
        self.value = value

unfenced = UnfencedStorage()
unfenced.write("value-from-B")
unfenced.write("value-from-A (stale)")  # A's late write silently clobbers B's
print(f"final stored value without fencing: {unfenced.value!r} (B's committed write was lost)")

Output:

leader B (token=34) write accepted: True, storage value now: 'value-from-B'
leader A (token=33) write accepted: False
storage's highest_token_seen: 34
final stored value: 'value-from-B'
rejected writes: [('leader-A', 33, 'value-from-A (stale)')]

-- for contrast, the SAME scenario WITHOUT fencing (storage accepts any write, no token check) --
final stored value without fencing: 'value-from-A (stale)' (B's committed write was lost)

Without the token check, leader A's stale write silently overwrites leader B's already-committed value: an invisible data loss, because nothing in the system detected a conflict, it just applied the last write it received. With the token check, storage rejects A's write outright because its token (33) is lower than the highest token it has already accepted (34). Note what the fencing token did NOT need: no network partition detection logic on A's part, no requirement that A know it lost its lease. The check lives entirely on the storage side and requires only a single integer comparison.

Trade-offs and pitfalls

• Quorum enforcement costs availability during a partition, by design: the minority side cannot write. That is the intended trade against consistency, not a bug, but it needs to be communicated to whoever is used to a system that stays fully available.
• Fencing tokens require every write path to actually carry and check the token. A single code path that writes directly to storage without going through the fencing check reopens the exact hole the token was meant to close; this is a common way the mechanism quietly stops protecting anything after a refactor.
• Automated reconciliation should be scoped narrowly. Fully automatic merge across arbitrary conflicting writes is where real incidents happen: a merge function that is correct for the common case can produce a plausible-looking but wrong result for an edge case nobody tested, and because it is automatic, nobody reviews it before it ships. Log every conflict that gets auto-resolved so it is auditable after the fact, even when you are confident in the merge logic.
• Detection has a real lag. Quorum loss on the minority side is detectable in roughly one heartbeat interval; a slow leader clinging to a stale lease can take up to the full lease timeout to get fenced. Set that timeout as a deliberate trade-off between failover speed and false-positive fencing during a merely slow (not actually dead) leader.

Direct answer

For a large, ever-growing events table, I would partition by date (typically monthly or weekly, sized so each partition stays in the tens-of-millions-of-rows range rather than billions) because almost every query and every retention decision on an events table is time-scoped, and date-range partitioning lets the planner skip partitions that fall entirely outside a query's time filter. That skipping is called partition pruning, and for a query targeting a recent window it turns a scan of the whole table into a scan of just the one or two partitions that overlap the requested range.

Structured elaboration

Partitioning strategies, briefly, since date is not the only option:

• Range partitioning (what I'm recommending here): each partition owns a contiguous range of values, e.g. occurred_at >= '2026-09-01' AND occurred_at < '2026-10-01'. Natural fit for time-series and for anything with a meaningful ordering (IDs, amounts).
• List partitioning: each partition owns an explicit set of discrete values, e.g. one partition per region IN ('US'), another for ('DE','FR','IT'). Fits categorical columns with a small, known set of values.
• Hash partitioning: the database hashes the partition key and assigns the row by hash bucket. Fits a column with no natural ranges or categories (like a user_id) where the goal is even distribution rather than query-driven pruning; the trade-off is that hash partitions only support pruning for equality predicates, not ranges, since a range of hash values has no relationship to a range of key values.

How pruning works and why it matters for recent-window queries. The planner compares the query's WHERE clause against each partition's boundary before deciding which partitions to even open. A query for "events from the last 7 days" against monthly partitions typically prunes to 1 (or 2, at a month boundary) partitions out of however many exist; a query for "events in September" prunes to exactly the September partition. This is the entire performance benefit of date partitioning: without it, every query pays for a full-table index scan (or worse, a sequential scan) across years of history to find a week of rows.

Operational considerations for adding and dropping partitions:

• Adding: create next month's (or week's) partition ahead of when it is needed, either via a scheduled job or a pre-created rolling window (e.g. always keep the next 3 months created). Creating a partition is a fast metadata operation; forgetting to create one before it's needed means inserts for that range either fail or, in engines with a default/catch-all partition, silently land in the wrong place.
• Dropping / retention: detach the oldest partition and drop it, instead of DELETE FROM events WHERE occurred_at < .... A DELETE has to find and mark every matching row individually, generates a write-ahead-log (WAL) entry, a durable, ordered record of the change written before it's applied, per row, and leaves behind dead space that autovacuum (Postgres's background process that reclaims that dead space automatically) then has to reclaim. Detaching a partition and dropping the now-standalone table is closer to an instant metadata change: no per-row scan, no bloat (that same leftover dead space) cleanup needed on the data that's gone.
• Sub-partitioning for a second access dimension. A 10-billion-row events table partitioned by date but also frequently filtered by user_id benefits from a two-level scheme: range-partition by month, then hash-sub-partition each month's partition by user_id. Date-range queries still prune to the right month(s); a query for one user within a date range additionally prunes to that user's hash bucket within the relevant month, instead of scanning the whole month for one user.
• When partitioning stops being enough and sharding becomes necessary. Partitioning helps as long as the whole table, across all its partitions, still fits comfortably on one instance's disk and the total write/read throughput fits on one instance's CPU and I/O. The concrete signals that it's time to shard instead: total data size approaching the instance's practical storage ceiling, write throughput saturating a single primary regardless of how well individual partitions are indexed, or per-partition maintenance (vacuum, index rebuilds) starting to compete for the same shared I/O budget across all partitions on the one machine. Partitioning is a single-instance technique; once the instance itself is the bottleneck rather than any one query, only spreading across instances (sharding) fixes it.

Worked example

CREATE TABLE evt (id BIGSERIAL, occurred_at DATE NOT NULL, payload TEXT)
  PARTITION BY RANGE (occurred_at);
CREATE TABLE evt_2026_07 PARTITION OF evt FOR VALUES FROM ('2026-07-01') TO ('2026-08-01');
CREATE TABLE evt_2026_08 PARTITION OF evt FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');
CREATE TABLE evt_2026_09 PARTITION OF evt FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');
-- 90,000 rows seeded across Jul-Sep 2026

Pruning to a single day (still lands entirely in one month's partition):

EXPLAIN (COSTS OFF) SELECT count(*) FROM evt WHERE occurred_at = '2026-09-15';
 Aggregate
   ->  Seq Scan on evt_2026_09 evt
         Filter: (occurred_at = '2026-09-15'::date)

Pruning to a range spanning two months (exactly those two, not all three):

EXPLAIN (COSTS OFF) SELECT count(*) FROM evt
WHERE occurred_at >= '2026-08-20' AND occurred_at < '2026-09-05';
 Aggregate
   ->  Append
         ->  Seq Scan on evt_2026_08 evt_1  Filter: (occurred_at >= '2026-08-20' AND occurred_at < '2026-09-05')
         ->  Seq Scan on evt_2026_09 evt_2  Filter: (occurred_at >= '2026-08-20' AND occurred_at < '2026-09-05')

evt_2026_07 never appears in either plan; it was pruned entirely. Retention via detach, instead of a row-by-row delete:

ALTER TABLE evt DETACH PARTITION evt_2026_07;
-- evt_2026_07 is now a standalone table; DROP TABLE evt_2026_07 frees the space
-- immediately, vs DELETE FROM evt WHERE occurred_at < '2026-08-01' which has to
-- scan and mark every matching row, write WAL for each, and needs a VACUUM after.

And list partitioning for a discrete-value example, pruning to the one matching partition:

EXPLAIN (COSTS OFF) SELECT * FROM evt_by_region WHERE region = 'DE';
 Seq Scan on evt_region_eu evt_by_region
   Filter: (region = 'DE')

(evt_region_eu was declared FOR VALUES IN ('DE','FR','IT'); a third partition, evt_region_default, catches any region not explicitly listed, e.g. 'JP' landed there in this run rather than causing an insert error.)

Trade-offs & pitfalls

• Too many partitions is a real failure mode, not just a theoretical one. Every partition is a separate object in the database's catalog; a planner has to consider every partition's bounds during planning, and operations like VACUUM and autovacuum run per partition. Daily partitions on a table kept for 10 years is 3,650+ partitions, and catalog and planning overhead starts to show up as a real cost, on top of the operational burden of thousands of tiny maintenance jobs. Size partitions for the query pattern (weekly or monthly is usually right for a few years of retention) rather than defaulting to the finest possible grain.
• Uneven partition sizes undermine the whole point. If traffic to an events table triples during a launch month, that month's partition can dwarf the others, and a query scoped to that one "pruned" partition is no faster than an unpartitioned table would have been for that period. Partitioning does not remove the need to think about skew; it just changes where the skew shows up.
• Indexing effects are easy to miss. Indexes in most partitioned engines are local to each partition by default, not global. A UNIQUE constraint that must hold across the whole table (not just within one partition) generally has to include the partition key in the uniqueness check, which can force a schema compromise (e.g. requiring occurred_at to be part of a uniqueness constraint that conceptually shouldn't need it) purely to satisfy the partitioning implementation.
• Add-a-clustering-key as a complementary, not competing, technique for large analytical fact tables. On a 50-billion-row fact table used for BI aggregation, date-range partitioning handles coarse pruning, but ordering the rows within each partition by a clustering/sort key that matches common aggregation groupings (e.g. CLUSTER-ing a partition by region, product_id in Postgres, or an equivalent clustering key in a columnar warehouse) further reduces the bytes scanned per query by improving locality inside the already-pruned partition.

Direct answer

Database federation queries across multiple databases that each remain independently owned, operated, and never redistributed, typically via a federation layer that fans a query out to the relevant source databases and stitches the results together. Sharding, by contrast, deliberately redistributes ONE logical dataset across multiple nodes that a single system controls end to end. The difference that matters in practice: sharding is something you design and own, choosing the shard key and rebalancing strategy yourself; federation is something you do TO data you often do not own or cannot move, because it already lives in separately operated systems (a different team's service, an acquired company's legacy database, a third-party system you only have read access to).

When federation is the right call

A company acquires another company whose customer data lives in its own separate, actively-used production database, still owned and operated by that team, with its own release cadence and its own operational constraints. Migrating that data into a shared, sharded schema is a large, risky project with no urgent deadline forcing it. Federation lets a unified "search across all customers" feature query both databases live and merge the results, without either team giving up ownership of their own system or taking on a risky migration before there is a clear business reason to. This is the right trade when the data's owners, schemas, and operational lifecycles are genuinely going to stay independent for the foreseeable future, and the query patterns across them are occasional cross-source lookups rather than the bulk of the workload.

When federation introduces unacceptable complexity

A query that needs to join large amounts of data across two of the federated sources (a report needing every acquired company's transactions joined against the main company's customer table, filtered and sorted across the combined set) turns into pulling large result sets out of both systems over the network and joining them in application code or a federation middleware layer, since neither underlying database can push down a join against data it does not have. This scales badly and gets slower exactly as both datasets grow, with no way to add an index that spans both sides of the join. At that point, either the query pattern needs to change (accept eventually-consistent denormalized copies, replicated via change-data-capture (CDC: a technique that watches a source database's transaction log for every row-level insert, update, and delete, then streams those changes out as events, instead of querying the source live for each one) into one place purpose-built for the cross-source query, rather than querying live), or the underlying decision to keep the data federated instead of migrated needs to be revisited, since federation was never designed to make heavy cross-source joins fast.

Trade-offs and pitfalls

• Federation trades query performance and consistency for organizational and operational independence. It is the right tool when that independence is the actual constraint (you cannot or should not move the data), not a substitute for sharding when you simply have not gotten around to a proper data architecture.
• A federation layer becomes a hidden single point of failure and a hidden performance bottleneck if every cross-source query has to flow through it, even though no single underlying database is overloaded.
• Schema drift across the federated sources is a real ongoing cost: each source can and will change its own schema independently, on its own schedule, and the federation layer has to keep working across those changes without any coordinated migration process to keep everything in sync.

Direct answer

A consistent point-in-time snapshot across shards without stopping writes comes from agreeing on a global "as of" marker, either a coordinator-broadcast epoch or a bounded, synchronized clock, and having every shard independently produce its OWN local MVCC snapshot as of that marker, then trusting that the union of those independently-taken local snapshots forms one globally consistent view, without ever pausing writes on any shard to take it. The two dominant approaches are a coordinator that broadcasts a fence (an epoch boundary) and waits for every shard to acknowledge it has flushed everything before that boundary, or a globally synchronized clock precise enough that a single timestamp, chosen once, is meaningful across every shard without any message round trip at all.

Why this is possible without stopping writes

Multi-Version Concurrency Control (MVCC), the mechanism most modern relational databases already use for ordinary transaction isolation, is the local building block: instead of overwriting a row in place, a write creates a new version and keeps the old one around (tagged with the transaction ID that created it and, once superseded, the one that replaced it), so a reader can ask for "the version of this row as it existed at transaction T" and get a consistent answer without blocking any concurrent writer. A distributed snapshot generalizes this single-node trick: get every shard to independently answer "what did MY data look like as of marker E," where E is agreed upon globally, and the combination of those per-shard answers is itself a consistent cut, even though no single moment of true global simultaneity ever had to be enforced by pausing anything.

Structured elaboration

Coordinator-driven epoch fencing

1. The coordinator picks the next epoch number, E, and broadcasts a fence: "every commit from now on is tagged E+1 or later."
2. Every shard, on receiving the fence, finishes applying any commit already in flight that was tagged E or earlier, then begins tagging new commits E+1.
3. Once every shard acknowledges "I have applied everything tagged E and nothing tagged E will arrive late," the coordinator declares epoch E closed; each shard's local MVCC snapshot at that point (everything up to and including epoch E, nothing after) is that shard's contribution to the global snapshot.
4. No shard ever stopped accepting writes; it only had to briefly track which epoch each write belonged to and confirm when it had fully drained the closing one.

Clock-based approaches

Instead of a message-driven fence, a system with a sufficiently precise, bounded-uncertainty global clock can skip the round trip entirely: assign every commit a timestamp from that clock, and a snapshot read at timestamp T returns the most recent version of every row committed before T, executed independently on each shard with no coordination message needed at read time. Google Spanner's TrueTime is the well-known real-world example of this approach: it exposes clock uncertainty as an explicit bounded interval rather than a single value, and uses a "commit wait" step (a transaction does not report itself committed until its chosen timestamp is guaranteed to be in the past across the whole uncertainty bound) specifically so that a snapshot read at a given timestamp is guaranteed not to miss any transaction that genuinely committed before it, all without needing to contact every other node at read time (Google Cloud documentation, "Spanner: TrueTime and external consistency"). The trade-off against the fencing approach: it needs specialized, tightly bounded clock synchronization infrastructure, not just ordinary NTP, but in exchange it removes the coordination round trip from every snapshot read, not just from the write path.

What "consistent" means here, precisely

The resulting cut is consistent in the sense that matters for backup and analytics: no row appears twice (once in its old version, once in its new one) and no committed write is silently missing from every shard's contribution, because the epoch boundary or timestamp draws a single, agreed line that both the fencing protocol and the clock-based approach enforce independently on each shard. It does not mean every shard's local time or state was literally identical at one physical instant, only that the SET of commits included is the same one a true instantaneous freeze would have captured.

Worked example

The single-shard MVCC mechanics this whole approach builds on, executed against a real PostgreSQL 16 instance, which already implements per-row versioning internally:

CREATE TABLE transactions (transaction_id BIGINT PRIMARY KEY, amount NUMERIC(12,2));
INSERT INTO transactions VALUES (1, 19.99);

SELECT txid_current();
SELECT xmin, xmax, transaction_id, amount FROM transactions LIMIT 1;
SELECT pg_current_wal_lsn();

Output:

 txid_current
--------------
          758
(1 row)

 xmin | xmax | transaction_id | amount
------+------+----------------+--------
  757 |    0 |              1 |  19.99
(1 row)

 pg_current_wal_lsn
--------------------
 0/19CF188
(1 row)

(The exact txid_current/xmin/LSN values above are specific to this session's transaction history and will differ on any other run; what matters, and is deterministic given the SQL shown, is the RELATIONSHIP between them, that xmin on the freshly inserted row equals the transaction ID that inserted it, xmax is 0 until superseded, and pg_current_wal_lsn is monotonically increasing, not the specific numbers.)
xmin (757) is the ID of the transaction that created this row version, one less than txid_current()'s next value (758) because the insert's transaction already completed; xmax of 0 means no later transaction has superseded it yet. This confirms transactions are simply monotonically increasing numbers this database already tracks per-commit, exactly the kind of local, ordered marker a coordinator-driven epoch or a synchronized-clock timestamp needs each shard to have available locally; this demonstration shows that primitive is not hypothetical, it already exists inside a single Postgres instance. pg_current_wal_lsn() shows the analogous idea for the write-ahead log itself: a monotonically increasing position that can serve as a shard's own local "as of" marker when correlating against a globally broadcast epoch. Extending this to a distributed snapshot means having a coordinator (or a synchronized clock) hand every shard the SAME target marker and having each shard independently answer "give me the MVCC-visible state as of that marker," which is a straightforward generalization of the single-node query above, not a fundamentally different mechanism.

Trade-offs and pitfalls

• A coordinator-driven fence adds real latency to establishing a NEW snapshot boundary (every shard must acknowledge before the epoch is declared closed) but adds zero cost to the ordinary write path in between snapshots; a clock-based approach adds a small, constant cost to every single write (commit-wait) but zero coordination cost per snapshot request, the right choice depends on whether you take snapshots rarely (favor fencing) or need frequent, ad hoc snapshot reads at arbitrary times (favor a bounded clock).
• The clock-based approach's safety guarantee is only as good as the clock's actual bound; a system that claims tight clock synchronization but does not actually enforce or monitor that bound (ordinary NTP without a hardware-backed bound, for instance) can silently violate the "no missing commit" property under real clock drift.
• A shard that is partitioned away and cannot acknowledge the fence blocks the ENTIRE snapshot from closing under the coordinator-driven approach; decide explicitly whether a snapshot should wait indefinitely for a slow or unreachable shard, or proceed and mark that shard's contribution as unavailable, since "wait forever" turns an unrelated shard's partition into an availability problem for backups and analytics on every OTHER shard too.
• Confusing "consistent as of a marker" with "identical to the live, current state" is a common misreading; a snapshot is deliberately a slightly-in-the-past, frozen view by design, and any process consuming it needs to be built to accept that staleness, not surprised by it.