From f8b875169a53244534f457f4d120940258caddde Mon Sep 17 00:00:00 2001 From: michael stack Date: Sun, 19 Jul 2026 17:31:45 -0700 Subject: [PATCH 1/2] DD: config-driven bounded-time rollback of shard-encoded location metadata Rolling SHARD_ENCODE_LOCATION_METADATA true->false could not reach `audit_storage metadata_encoding` ROLLBACK COMPLETE in bounded time: converged and "safe to downgrade binary" was unverifiable short of a storage wiggle (hours-to-days). This adds an opt-in, bounded-time DD-init rewrite that converges existing keyServers AND serverKeys back to old format and clears DataMoveMetaData, driven by two new DatabaseConfiguration options: configure shard_metadata_format = original | encoded (target format) configure shard_metadata_migration = enabled | disabled (run rewrite) Config is the source of truth for DD's encoding target, resolved once at init as `shard_metadata_format ?? SHARD_ENCODE_LOCATION_METADATA` (the knob is only the fallback when the config is UNSET) and used across all DD write/move paths -- so a rollback is driven by `configure shard_metadata_format=original` with no knob flip and no process restart (the configure triggers a recovery that re-inits DD). Deploying the binary is a byte-identical no-op until an operator opts in. A completion sentinel records ROLLBACK COMPLETE and fast-paths later inits. serverKeys are rewritten serially per storage server, so rewrite time scales with cluster size (minutes typically); progress is visible via DDShardEncode* trace events and the audit. Caveats (details in design/shard-encode-location-metadata.md and documentation/sphinx/source/command-line-interface.rst): - Binary downgrade: a pre-config binary reads only the knob, so downgrade is safe only at ROLLBACK COMPLETE with the target binary deployed SHARD_ENCODE_LOCATION_METADATA=false. - Known limitation: large teams (ddLargeTeamEnabled) and physical shard (ENABLE_DD_PHYSICAL_SHARD), mutually exclusive with shard encoding, still read the raw knob, not the config. Once shard_metadata_format is set, don't move the knob except for a downgrade. Tested: sim tests/fast/ShardEncodeRollback.toml (knob-fallback path) and tests/fast/ShardEncodeRollbackConfig.toml (config-only, no knob flip), each passing a 100,000-run joshua ensemble; plus a live-cluster forward->rollback->re-forward under mako load with DD-pod kills (test_shardencode_rollover_load, fdb-kubernetes-tests). Builds on #13310 (SHARD_ENCODE_LOCATION_METADATA audit + rollback support). Depends on #13670 (audit_storage counting fix, merged) to observe ROLLBACK COMPLETE. --- design/shard-encode-location-metadata.md | 304 ++++++++-- .../sphinx/source/command-line-interface.rst | 167 ++++- fdbcli/CheckMetadataEncodingCommand.cpp | 12 +- fdbcli/ConfigureCommand.cpp | 15 +- fdbclient/DatabaseConfiguration.cpp | 41 +- fdbclient/ManagementAPI.cpp | 18 + fdbclient/Schemas.cpp | 10 + fdbclient/SystemData.cpp | 8 + .../include/fdbclient/DatabaseConfiguration.h | 43 ++ fdbclient/include/fdbclient/SystemData.h | 26 + .../clustercontroller/ClusterRecovery.cpp | 6 +- fdbserver/core/MoveKeys.cpp | 92 ++- fdbserver/core/SeedShardServers.cpp | 7 +- fdbserver/core/ServerKnobs.cpp | 1 + fdbserver/core/include/fdbserver/core/Knobs.h | 12 +- .../core/include/fdbserver/core/MoveKeys.h | 25 + .../include/fdbserver/core/SeedShardServers.h | 5 +- .../datadistributor/DDRelocationQueue.cpp | 22 +- fdbserver/datadistributor/DDShardTracker.cpp | 9 +- fdbserver/datadistributor/DDShardTracker.h | 7 + .../DDTeamCollection.actor.cpp | 3 +- fdbserver/datadistributor/DDTxnProcessor.cpp | 573 ++++++++++++++++-- fdbserver/datadistributor/DDTxnProcessor.h | 6 +- .../datadistributor/DataDistribution.cpp | 43 +- fdbserver/datadistributor/DataDistribution.h | 4 +- fdbserver/workloads/CheckMetadataEncoding.cpp | 310 +++++++--- tests/CMakeLists.txt | 1 + tests/fast/ShardEncodeRollback.toml | 93 ++- tests/fast/ShardEncodeRollbackConfig.toml | 118 ++++ 29 files changed, 1696 insertions(+), 285 deletions(-) create mode 100644 tests/fast/ShardEncodeRollbackConfig.toml diff --git a/design/shard-encode-location-metadata.md b/design/shard-encode-location-metadata.md index cd1821bcee3..eb31069e346 100644 --- a/design/shard-encode-location-metadata.md +++ b/design/shard-encode-location-metadata.md @@ -317,31 +317,130 @@ shard IDs on every entry (auditing, bulk load). Basic DD operation works without ## Rollout and Rollback -### Rollout (enable knob) +### Rollout (enable new format) Safe — the new binary reads both formats. Mixed-format coexistence works because -the decoder checks the protocol version embedded in each value. +the decoder checks the protocol version embedded in each value. Enable via +`configure shard_metadata_format=encoded` (or, on a cluster that has not set +the config, the `SHARD_ENCODE_LOCATION_METADATA=true` knob fallback). + +### Rollback within same binary (config-driven) + +Safe — tested and verified in simulation (`ShardEncodeRollback.toml` for the +knob-fallback path, `ShardEncodeRollbackConfig.toml` for the config-driven +path). + +**Migration is optional.** Setting the target back to old format +(`configure shard_metadata_format=original`, no restart) is a +zero-additional-work operation from the cluster's perspective: + +- The cluster keeps running fine after the target change. DD's decoders + handle both old and new formats; SS's `applyPrivateData` accepts + both. +- DD writes old-format for every new shard move going forward. +- Existing new-format entries linger. `audit_storage + metadata_encoding` reports `MIGRATION IN PROGRESS` indefinitely. + +The only thing this mixed state prevents is **downgrading the FDB +binary**. Old FDB versions cannot decode new-format `serverKeys` +values and will misbehave. To make downgrade safe, all entries must +be drained to old format — reflected in `audit_storage +metadata_encoding` returning `ROLLBACK COMPLETE — safe to downgrade +binary`. + +Operators pick from three paths to reach `ROLLBACK COMPLETE`: + +1. **Do nothing beyond setting the target.** Natural DD moves gradually + drain entries over time. Timeframe unbounded — could be days or + weeks on a quiescent cluster. Fine if you don't need to downgrade + on a schedule. + +2. **Active rewrite (this doc's remaining subsections).** Opt in via + the two DatabaseConfiguration options. Bounded — scales with cluster + size (the serverKeys drain is serial per SS): minutes on typical + clusters, potentially hours on a very large (~250k-shard) cluster. + These are estimates. Progress is observable via DD trace events and + `audit_storage metadata_encoding` (see the serverKeys subsection). + +3. **Storage wiggle.** Enable perpetual storage wiggle. Every SS + rotates through the cluster, causing every shard to move at least + once. Slow (hours-to-days) but exercises only pre-existing code + paths. Recommended when active rewrite is too aggressive for + safety envelope. + +The active rewrite path described below is **opt-in per operator event** +via two database configuration options that mirror the +`storage_engine` + `perpetual_storage_wiggle` pattern: -### Rollback within same binary (flip knob false) +``` +fdbcli> configure shard_metadata_format=original +fdbcli> configure shard_metadata_migration=enabled +``` + +- `shard_metadata_format` = the target format DD converges existing + entries toward. When unset, DD falls back to the + `SHARD_ENCODE_LOCATION_METADATA` knob (`true → encoded`, + `false → original`). +- `shard_metadata_migration` = whether DD actively runs the rewrite + pass at init. When unset or `disabled` (default), DD does nothing at + init and both directions converge via natural DD moves or a manual + storage wiggle (unbounded timescale). -Safe — tested and verified in simulation (`ShardEncodeRollback.toml`): +Deploying a binary with this feature onto an existing cluster is a +byte-identical no-op unless the operator explicitly enables migration. + +When migration is enabled, the safety properties below apply: 1. **Decoders handle both formats.** SS processes both old-format and shard-encoded serverKeys mutations correctly regardless of its persistent `shardAware` flag. No data loss from format mismatch. -2. **DD restarts on knob change.** In production, changing the knob requires a - process restart. On restart, DD initializes with the new value and runs the - startup rewrite. In-flight actors from the old DD are cancelled. - -3. **DD rewrites keyServers on startup.** When DD restarts with knob=false, it - scans keyServers entries and rewrites shard-encoded ones to old tag-based format. - DataMoveMetaData entries are cleared. - -4. **serverKeys drains naturally.** Shard-encoded serverKeys entries are NOT - rewritten on startup (doing so causes KRM fragmentation and SS pair-processing - issues). They remain readable in both formats and drain to old format as DD - moves shards using the old path. +2. **DD re-inits on a config change.** A `configure shard_metadata_format=...` + (or `shard_metadata_migration=...`) change triggers a recovery that + re-elects DD; the new DD resolves the effective target and runs the + startup rewrite. No knob change or process restart is required (the knob + is only the fallback when the config is unset). In-flight actors from the + old DD are cancelled. + +3. **DD rewrites keyServers on startup.** When DD re-inits with the effective + target = `original` and migration=enabled, it scans keyServers entries and + rewrites shard-encoded ones to old tag-based format. DataMoveMetaData + entries are cleared. + +4. **serverKeys are rewritten on startup.** DD scans each SS's serverKeys + KRM and rewrites shard-encoded ranges back to old-format constants + using `krmSetRangeCoalescing` — the same primitive natural moves + use. This preserves the two KRM invariants (no fragmentation because + coalescing merges adjacent same-value entries; paired `[begin, end)` + mutations because that is what `krmSetRangeCoalescing` emits, matching + what `applyPrivateData` on the SS expects). Unlike the keyServers + rewrite (step 3), the serverKeys rewrite is not paginated across + DD-init re-invocations: a single invocation drains every SS's + serverKeys KRM (looping over the serverList until it stabilizes), + then writes the completion sentinel. It blocks DD init for the + duration of the drain, and the serverKeys rewrite processes storage + servers **serially** (``co_await`` per SS, each SS re-scanned until a + pass rewrites nothing), so the time scales with cluster size: + (# SSes) × (shard-encoded ranges per SS). On typical clusters this is + minutes (e.g. a 12-SS live-cluster test drained within minutes); on a + very large (~250k-shard) cluster it could be substantially longer — + order of hours. These are estimates, not measured bounds. It is a + rare, opt-in, per-rollback (``shard_metadata_migration=enabled``) + event. + + Progress is observable while it runs: DD emits ``DDShardEncodeRewriteBegin``, + per-SS ``DDShardEncodeRollbackPhase3SSStart`` / ``...Phase3SSDone``, and + ``DDShardEncodeRewriteComplete`` (with ``Phase3Rewrites`` / ``Phase3Passes`` + / ``Phase3ElapsedSec``) trace events; operators can also poll + ``fdbcli> audit_storage metadata_encoding`` (new-format counts trending to + zero) and ``fdbcli> location_metadata physicalshards`` (physical shards + trending to zero). + + Direct + `tr.set()` at each entry — an earlier design that this section used + to warn against — would produce fragmentation and violate the SS + pair-processing invariant; the current implementation avoids both + by routing through the same KRM primitive natural moves use. 5. **Safety net for rolling restarts.** During a rolling restart, an old DD instance (not yet killed) may find its DataMoveMetaData cleared by a new DD @@ -350,28 +449,153 @@ Safe — tested and verified in simulation (`ShardEncodeRollback.toml`): once more and pick up the new knob value. This is at most one extra restart per DD instance during the transition, not per-shard. -The rollback procedure: -1. Set `SHARD_ENCODE_LOCATION_METADATA=false` -2. Restart the `fdbserver` processes (knob change requires restart) -3. On DD init, keyServers entries are rewritten to old format and DataMoveMetaData is cleared -4. DD proceeds with old path for all new moves -5. serverKeys entries drain to old format as DD touches shards over time +6. **Sentinel-based idempotency.** DD writes a completion sentinel key + (`\xff/dd/shard_encode_migration_complete = "old"`) at the end of a + successful rewrite pass. Subsequent DD inits read the sentinel and + fast-path skip — steady-state cost when migration is enabled is one + key read per DD init. On forward-direction knob flips + (`shard_metadata_format=encoded` or knob=true), DD clears the + stale sentinel so subsequent rollback events don't fast-path skip + incorrectly. If DD dies mid-rewrite, the sentinel remains absent → + the next DD does a full scan and converges. + +The rollback procedure (config-driven, no knob flip, no restart): +1. `fdbcli> configure shard_metadata_format=original shard_metadata_migration=enabled` +2. Force DD to re-init so it picks up the new configuration + immediately (the configure in step 1 already triggers a + recovery; this only expedites it): + + fdbcli> datadistribution off + fdbcli> datadistribution on + + The newly-elected DD resolves `effective = original` from the + config and runs the rewrite pass. No process restart and no knob + flip is needed — DD's write paths follow the config target. +3. On the DD's next init, keyServers entries and serverKeys entries + are both rewritten to old format, and DataMoveMetaData is + cleared. The keyServers rewrite is paginated (up to 1000 entries + per DD-init pass, re-invoking until the prefix is covered); the + serverKeys rewrite then drains all storage servers within a + single DD-init invocation before the completion sentinel is + written. +4. DD proceeds with old path for all new moves. +5. Verify via `fdbcli> audit_storage metadata_encoding` returning + `ROLLBACK COMPLETE`. + +Alternative rollback procedure (drain via storage wiggle, no +configure needed): +1. Set `SHARD_ENCODE_LOCATION_METADATA=false` and restart processes. +2. Leave `shard_metadata_migration` unset or `disabled`. +3. Enable perpetual storage wiggle. Every SS rotates through, + causing DD to rewrite every shard's metadata in the current + (old) format as part of natural moves. +4. Verify via `audit_storage metadata_encoding`. Reaches + `ROLLBACK COMPLETE` over hours-to-days depending on cluster + size and wiggle throttling. + +Because DD rewrites serverKeys directly on init when migration is +enabled (using the same KRM primitive natural moves use — see step 4 +in the previous section), storage wiggle is not required to reach +`ROLLBACK COMPLETE` on a quiescent cluster. Both paths are +supported; operators pick based on how quickly they need +`ROLLBACK COMPLETE`. + +### Config is the source of truth; the knob is the fallback + +DD resolves its effective encoding target once at init: + +``` +effective = shard_metadata_format.present() + ? shard_metadata_format + : SHARD_ENCODE_LOCATION_METADATA // knob fallback +``` -No need to stop writes or trigger wiggle. +`shard_metadata_format` (configure option) is authoritative for **all** of +DD's decisions — both the migration decision at init and the write paths +(MoveKeys, natural moves, failed-server / dataMove cleanup). The +`SHARD_ENCODE_LOCATION_METADATA` knob is consulted **only as the fallback** +when `shard_metadata_format` is UNSET. + +DD does **not** write the config. It re-resolves the effective target on every +init, and a `configure shard_metadata_format=...` change triggers a recovery +that re-elects a DD which reads the new value. So nothing needs to be "kept in +sync" on a running cluster and there is no self-seeding write (which would +force an extra recovery on upgrade): just set the config and DD converges. +Deploying the binary is a byte-identical no-op until an operator sets the +config. + +**Two DD features stay gated on the raw knob, not the config target — a +known limitation.** Physical shard moves (`ENABLE_DD_PHYSICAL_SHARD`) and +per-range replication / "large teams" (`ddLargeTeamEnabled`, i.e. +`DD_MAX_SHARDS_ON_LARGE_TEAMS > 0 && !SHARD_ENCODE_LOCATION_METADATA`) are +**mutually exclusive with shard-encoded metadata** — they operate only in the +old (tag-based) format. They still read the `SHARD_ENCODE_LOCATION_METADATA` +knob directly rather than the resolved `shard_metadata_format` target. That is +a deliberate trade-off, not a full solution: + +- Gating them on `shard_metadata_format` would be *worse*: the config flips + instantly but the metadata drains asynchronously, so they would switch on + the moment `shard_metadata_format=original` is set — while encoded entries + still exist — violating the mutual exclusion during the active rollback. + Gating on the knob (which only changes via a process restart) avoids that. +- But the knob gating is not fully safe either, once config and knob can + disagree. Two consequences: + - After a **config-only** rollback (config=old, knob still `true`), large + teams / physical shard stay **disabled** even at `ROLLBACK COMPLETE`. To + re-enable them, redeploy with `SHARD_ENCODE_LOCATION_METADATA=false` (the + same knob=false deploy a binary downgrade needs — see the downgrade + contract below). + - Conversely, a contradictory `SHARD_ENCODE_LOCATION_METADATA=false` flip + while `shard_metadata_format=encoded` would **wrongly enable** large + teams / physical shard on encoded metadata. +- Impact is low in practice: large teams is rarely used, physical shard is + experimental (off by default), and the bad case requires an operator + deliberately contradicting the config with the knob. The correct fix (gate + these on the fully-drained old-format state) is deferred. + +**Operational rule:** once you set `shard_metadata_format`, do not move the +`SHARD_ENCODE_LOCATION_METADATA` knob except as part of a downgrade +(knob=false on a fully rolled-back cluster). Config drives the *encoding*; the +knob gates the mutually-exclusive *old-format-only features*. + +Neither of these affects the keyServers/serverKeys encoding or the +rollback/downgrade contract. ### Rollback to old binary (downgrade) NOT safe if shard-encoded values exist. Old binary cannot parse new-format -serverKeys values. Requires a migration step (drain all entries to old format) -before downgrade. +serverKeys values. Requires DD's rewrite to complete first — verify via +`audit_storage metadata_encoding` reporting `ROLLBACK COMPLETE`. + +**Downgrade contract (config-only clusters).** A pre-config binary cannot read +`shard_metadata_format`; it decides encoding **solely from the +`SHARD_ENCODE_LOCATION_METADATA` knob**. Config-only convergence leaves the +knob untouched (it may still be `true`). So a binary downgrade is safe only +when **both**: + +1. the cluster is already in unencoded (old) metadata (`ROLLBACK COMPLETE`), and +2. the downgrade-target binary is deployed with `SHARD_ENCODE_LOCATION_METADATA=false` + in its command-line / knob config. + +Condition 2 is natural — a binary downgrade is itself a redeploy/restart, so +you set the knob on the target binary at that moment; no live knob change on +the running config-only binary is needed. If you skip it, the old binary boots +with `knob=true` and resumes new-format writes, re-corrupting the metadata the +config-driven rollback just drained. (Downgrades *among* config-aware binaries +are unaffected — they all read `shard_metadata_format`.) ### Migration for downgrade -1. Set knob false, restart (DD uses old path, writes old format) -2. Run background rewriter or trigger storage wiggle to touch all shards -3. Verify completion: scan keyServers/serverKeys, confirm zero shard-encoded entries -4. Clear `\xff/dataMoves/` range -5. Safe to downgrade binary +1. Converge to old format via config, no restart: + `fdbcli> configure shard_metadata_format=original shard_metadata_migration=enabled` + (optionally `datadistribution off; on` to expedite). DD rewrites both + keyServers and serverKeys to old format on init; DataMoveMetaData is cleared. +2. Verify completion: `fdbcli> audit_storage metadata_encoding` reports + `ROLLBACK COMPLETE — safe to downgrade binary`. Large clusters may + require several DD init passes to fully drain; the tool's status + line shows progress. +3. Downgrade the binary, deploying it with `SHARD_ENCODE_LOCATION_METADATA=false` + (see the downgrade contract above). ### Verifying migration state @@ -437,29 +661,29 @@ both forward migration completion and rollback readiness. Runs client-side ``` fdbcli> audit_storage metadata_encoding keyServers: 45021 entries - Old format (tag-based): 0 - New format (UID-based): 45021 + Original format (tag-based): 0 + Encoded format (UID-based): 45021 serverKeys: 123004 entries - Old format (constants): 0 - New format (UID-encoded): 123004 + Original format (constants): 0 + Encoded format (UID-encoded): 123004 dataMoves: 3 entries (in-progress moves) Migration status: FORWARD COMPLETE ``` -After rollback (knob flipped false, drain in progress): +After rollback (target set to `original`, drain in progress): ``` fdbcli> audit_storage metadata_encoding keyServers: 45021 entries - Old format (tag-based): 44800 - New format (UID-based): 221 + Original format (tag-based): 44800 + Encoded format (UID-based): 221 serverKeys: 123004 entries - Old format (constants): 122500 - New format (UID-encoded): 504 + Original format (constants): 122500 + Encoded format (UID-encoded): 504 dataMoves: 0 entries -Migration status: MIGRATION IN PROGRESS (mixed format: 221 new keyServers, 504 new serverKeys) +Migration status: MIGRATION IN PROGRESS (mixed format: 221 encoded keyServers, 504 encoded serverKeys) ``` When all zeros: diff --git a/documentation/sphinx/source/command-line-interface.rst b/documentation/sphinx/source/command-line-interface.rst index 390c271e33a..656292713c6 100644 --- a/documentation/sphinx/source/command-line-interface.rst +++ b/documentation/sphinx/source/command-line-interface.rst @@ -64,7 +64,7 @@ The ``commit`` command commits the current transaction. Any sets or clears execu configure --------- -The ``configure`` command changes the database configuration. Its syntax is ``configure [new|tss] [single|double|triple|three_data_hall|three_datacenter] [ssd|memory] [grv_proxies=] [commit_proxies=] [resolvers=] [logs=] [count=] [perpetual_storage_wiggle=] [perpetual_storage_wiggle_locality=<:|0>] [perpetual_storage_wiggle_engine=] [storage_migration_type={disabled|aggressive|gradual}]``. +The ``configure`` command changes the database configuration. Its syntax is ``configure [new|tss] [single|double|triple|three_data_hall|three_datacenter] [ssd|memory] [grv_proxies=] [commit_proxies=] [resolvers=] [logs=] [count=] [perpetual_storage_wiggle=] [perpetual_storage_wiggle_locality=<:|0>] [perpetual_storage_wiggle_engine=] [storage_migration_type={disabled|aggressive|gradual}] [shard_metadata_format={original|encoded}] [shard_metadata_migration={enabled|disabled}]``. The ``new`` option, if present, initializes a new database with the given configuration rather than changing the configuration of an existing one. When ``new`` is used, both a redundancy mode and a storage engine must be specified. @@ -131,6 +131,171 @@ The default is ``disabled``, which means changing the storage engine will not be ``aggressive`` tries to replace as many storages as it can at once, and will recruit a new storage server on the same process as the old one. This will be faster, but can potentially hit degraded performance or OOM with two storages on the same process. The main benefit over ``gradual`` is that this doesn't need to take one storage out of rotation, so it works for small or development clusters that have the same number of storage processes as the replication factor. Note that ``aggressive`` is not exclusive to running the perpetual wiggle. ``disabled`` means that if the storage engine is changed, fdb will not move the cluster over to the new storage engine. This will disable the perpetual wiggle from rewriting storage files. +shard metadata format and migration +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Two related configuration options that control migration of the cluster's +shard-location metadata (``\xff/keyServers/`` and ``\xff/serverKeys/``) +between the legacy tag-based encoding and the newer UID+dataMoveId +encoding introduced with ``SHARD_ENCODE_LOCATION_METADATA``. The pair +mirrors the ``storage_engine`` + ``perpetual_storage_wiggle`` pattern. + +``shard_metadata_format={original|encoded}`` — target encoding +Data Distributor should converge existing metadata entries toward. When +unset, DD falls back to the legacy ``SHARD_ENCODE_LOCATION_METADATA`` +knob (``true → encoded``, ``false → original``). + +``shard_metadata_migration={enabled|disabled}`` — whether DD actively +runs a rewrite pass at init to converge existing entries to +``shard_metadata_format``. Default is unset (equivalent to +``disabled``); DD does nothing at init in either direction and entries +drain via natural DD moves or a manual storage wiggle (unbounded +timescale). Enable per-event when a bounded-time migration is needed. + +These options are how you drive a shard-metadata migration. If you never +set them, DD falls back to the ``SHARD_ENCODE_LOCATION_METADATA`` knob for +its target format and runs no active rewrite. See "Do I need to migrate?" +below for the decision. + +Do I need to migrate? +~~~~~~~~~~~~~~~~~~~~~ + +**No, unless you plan to downgrade the FDB binary.** After rolling the +target to old format (``configure shard_metadata_format=original``, or — +with the config unset — the legacy ``SHARD_ENCODE_LOCATION_METADATA=false`` +knob), the cluster keeps running normally: DD writes old-format for every +new shard move, existing new-format entries linger, ``audit_storage +metadata_encoding`` reports ``MIGRATION IN PROGRESS`` indefinitely. +This mixed state is fully functional. The current binary understands +both formats. + +The one thing you *cannot* do while entries are in mixed state is +downgrade to an older FDB binary. Old binaries cannot parse +new-format ``serverKeys`` entries and will misbehave. To make +downgrade safe, all new-format entries must be rewritten to old +format — reflected in ``audit_storage metadata_encoding`` returning +``ROLLBACK COMPLETE — safe to downgrade binary``. + +Three ways to reach ``ROLLBACK COMPLETE``: + +**A. Do nothing beyond setting the target.** Rely on natural DD moves +over time. Every shard move rewrites its metadata in the current +(now old) format. Eventually every entry drains. **Timeframe is +unbounded** and depends on workload — could be days or weeks on a +quiescent cluster; hours on a busy one. If you don't have a specific +downgrade deadline, this is the least effort. + +**B. Active rewrite (this PR's fast path).** Set the two configure +options plus toggle DD. Bounded, but the time scales with cluster size — +the serverKeys drain runs serially per storage server: minutes on typical +clusters, potentially hours on a very large (250k-shard) cluster (these +are estimates). Progress is observable (see the Rollback flow below). + +**C. Storage wiggle.** Enable ``perpetual_storage_wiggle=1``. Every +SS gets rotated, forcing every shard to move at least once. Slow +(hours to days depending on data size and wiggle throttling) but +proven — no new code paths involved. Recommended when the active +rewrite is too aggressive for your safety envelope. + +Rollback flow (drain new-format entries to old; bounded, but the time +scales with cluster size — the serverKeys drain runs serially per storage +server, so minutes on typical clusters and potentially hours on a very +large 250k-shard cluster; these are estimates). This is driven purely by +configuration — no knob change and no process restart. DD's effective +target comes from ``shard_metadata_format``; the +``SHARD_ENCODE_LOCATION_METADATA`` knob is only a fallback consulted when +the config is unset, so the knob is not touched here: + +.. code-block:: console + + # 1. Set the target to original and enable migration (single + # fdbcli command). No knob flip, no rolling restart. + fdbcli> configure shard_metadata_format=original shard_metadata_migration=enabled + + # 2. The configure triggers a recovery, so the re-elected DD picks + # up the new target automatically. To expedite (optional), force a + # DD re-election — a few seconds, not a process restart: + fdbcli> datadistribution off + fdbcli> datadistribution on + + # 3. Watch progress. DD emits DDShardEncodeRewriteBegin, per-SS + # DDShardEncodeRollbackPhase3SSStart/SSDone, and + # DDShardEncodeRewriteComplete (Phase3Rewrites/Phase3Passes/ + # Phase3ElapsedSec) trace events. You can also poll: + fdbcli> audit_storage metadata_encoding + # (new-format counts trend to zero) and + fdbcli> location_metadata physicalshards + # (physical shards trend to zero) + # ... poll until "ROLLBACK COMPLETE — safe to downgrade binary" ... + + # 4. Optionally leave migration enabled (steady-state cost is + # one key read per DD init) or disable it: + fdbcli> configure shard_metadata_migration=disabled + +Downgrading the binary afterward: an older (pre-config) binary cannot +read ``shard_metadata_format`` and decides encoding solely from the +``SHARD_ENCODE_LOCATION_METADATA`` knob. A binary downgrade is therefore +safe only once ``ROLLBACK COMPLETE`` is reached **and** the +downgrade-target binary is deployed with +``SHARD_ENCODE_LOCATION_METADATA=false`` in its knob config. A downgrade +is itself a redeploy/restart, so you set the knob on the target binary at +that moment — no live knob change on the running cluster is needed. If you +skip it, the old binary boots with ``knob=true`` and resumes new-format +writes, re-corrupting the metadata the rollback just drained. + +Old-format-only features after a config-only rollback (known limitation): +per-range replication ("large teams", ``DD_MAX_SHARDS_ON_LARGE_TEAMS``) and +physical shard moves (``ENABLE_DD_PHYSICAL_SHARD``) are mutually exclusive +with shard-encoded metadata and still read the +``SHARD_ENCODE_LOCATION_METADATA`` knob directly, not ``shard_metadata_format``. +Consequences: after a config-only rollback (knob still true) they stay +disabled even once ``ROLLBACK COMPLETE`` is reached; and a contradictory +``SHARD_ENCODE_LOCATION_METADATA=false`` flip while +``shard_metadata_format=encoded`` would wrongly enable them on encoded +metadata. Gating them on ``shard_metadata_format`` instead would be worse +(the config flips instantly while metadata drains asynchronously, enabling +them mid-rollback while encoded entries still exist), so they are left on the +knob; impact is low (large teams is rarely used, physical shard is +experimental). Rule: once you set ``shard_metadata_format``, do not move the +knob except as part of a downgrade. To re-enable these features on a +rolled-back cluster, redeploy with ``SHARD_ENCODE_LOCATION_METADATA=false`` +(the same knob=false deploy a binary downgrade requires). + +Re-forward flow (after a rollback), also config-driven (no knob flip): + +.. code-block:: console + + # 1. Set the target back to encoded and (re-)enable migration: + fdbcli> configure shard_metadata_format=encoded shard_metadata_migration=enabled + + # 2. Force DD to re-init to pick up the new configuration (optional; + # the configure already triggers a recovery): + fdbcli> datadistribution off + fdbcli> datadistribution on + + # 3. The re-elected DD clears any stale rollback-complete + # sentinel. Old-format entries drain via natural DD moves; + # new-format entries appear as DD writes new moves. + # `FORWARD COMPLETE` converges over the natural-move + # timescale (unbounded) rather than in a bounded post-flip + # window. New-format code paths accept old-format entries, + # so mixed state is safe throughout. + fdbcli> audit_storage metadata_encoding + +Minimum recipe: if ``shard_metadata_format`` is left unset, DD falls +back to reading the ``SHARD_ENCODE_LOCATION_METADATA`` knob directly +for the target format. You can therefore skip setting the format +and just enable migration; DD infers direction from the knob: + +.. code-block:: console + + fdbcli> configure shard_metadata_migration=enabled + fdbcli> datadistribution off ; datadistribution on + +For more details including the safety and observability story, see +the ``design/shard-encode-location-metadata.md`` design document in +the source tree. + consistencyscan ---------------- diff --git a/fdbcli/CheckMetadataEncodingCommand.cpp b/fdbcli/CheckMetadataEncodingCommand.cpp index 246ab5d62a1..e7706b3786e 100644 --- a/fdbcli/CheckMetadataEncodingCommand.cpp +++ b/fdbcli/CheckMetadataEncodingCommand.cpp @@ -136,11 +136,11 @@ Future checkMetadataEncodingCommandActor(Database cx, std::vector checkMetadataEncodingCommandActor(Database cx, std::vector 0 || serverKeysNew > 0) { - fmt::println("Migration status: MIGRATION IN PROGRESS (mixed format: {} new keyServers, {} new serverKeys)", + fmt::println("Migration status: MIGRATION IN PROGRESS (mixed format: {} encoded keyServers, {} encoded serverKeys)", keyServersNew, serverKeysNew); } else { - fmt::println("Migration status: NOT STARTED (all old format)"); + fmt::println("Migration status: NOT STARTED (all original format)"); } co_return true; diff --git a/fdbcli/ConfigureCommand.cpp b/fdbcli/ConfigureCommand.cpp index 521289e0832..1d4012c656e 100644 --- a/fdbcli/ConfigureCommand.cpp +++ b/fdbcli/ConfigureCommand.cpp @@ -331,6 +331,8 @@ void configureGenerator(const char* text, "perpetual_storage_wiggle_locality=", "perpetual_storage_wiggle_engine=", "storage_migration_type=", + "shard_metadata_format=", + "shard_metadata_migration=", nullptr }; arrayGenerator(text, line, opts, lc); } @@ -343,7 +345,9 @@ CommandFactory configureFactory( "commit_proxies=|grv_proxies=|logs=|resolvers=>*|" "count=|perpetual_storage_wiggle=|perpetual_storage_wiggle_locality=" "<:|0>|perpetual_storage_wiggle_engine=|" - "storage_migration_type={disabled|gradual|aggressive}" + "storage_migration_type={disabled|gradual|aggressive}|" + "shard_metadata_format={original|encoded}|" + "shard_metadata_migration={enabled|disabled}" "|exclude=", "change the database configuration", "The `new' option, if present, initializes a new database with the given configuration rather than changing " @@ -375,6 +379,15 @@ CommandFactory configureFactory( "perpetual_storage_wiggle_locality=<:|0>: Set the process filter for wiggling. " "The processes that match the given locality key and locality value are only wiggled. The value 0 will disable " "the locality filter and matches all the processes for wiggling.\n\n" + "shard_metadata_format={original|encoded}: Target encoding for the cluster's shard-location metadata " + "(\\xff/keyServers/ and \\xff/serverKeys/). `encoded' is the shard-encoded (UID+dataMoveId) format, " + "`original' the legacy tag-based format. Paired with shard_metadata_migration below; mirrors the " + "storage_engine + perpetual_storage_wiggle pattern. When unset, DD falls back to the legacy " + "SHARD_ENCODE_LOCATION_METADATA knob (true -> encoded, false -> original).\n\n" + "shard_metadata_migration={enabled|disabled}: When enabled, DD at init actively converges existing " + "shard metadata entries to shard_metadata_format. When disabled (default), DD does nothing at init " + "in either direction; entries drain via natural DD moves or a manual storage wiggle. See the " + "SHARD_ENCODE_LOCATION_METADATA design doc for cost model and safety analysis.\n\n" "exclude=: Sets the addresses in the format of IP1:port1,IP2:port2 pairs to be excluded during " "recruitment. Note this should be only used when the database is unavailable because of the faulty processes " "that are blocking the recovery from completion. The number of addresses should be less than the replication " diff --git a/fdbclient/DatabaseConfiguration.cpp b/fdbclient/DatabaseConfiguration.cpp index f25072085d8..ce7f05cc4a1 100644 --- a/fdbclient/DatabaseConfiguration.cpp +++ b/fdbclient/DatabaseConfiguration.cpp @@ -56,6 +56,8 @@ void DatabaseConfiguration::resetInternal() { perpetualStorageWiggleSpeed = 0; perpetualStorageWiggleLocality = "0"; storageMigrationType = StorageMigrationType::DEFAULT; + shardMetadataFormat = ShardMetadataFormat::UNSET; + shardMetadataMigration = ShardMetadataMigration::UNSET; } int toInt(ValueRef const& v) { @@ -265,7 +267,15 @@ bool DatabaseConfiguration::isValid() const { // We cannot specify regions with three_datacenter replication LOG_TEST((perpetualStorageWiggleSpeed == 0 || perpetualStorageWiggleSpeed == 1)) && LOG_TEST(isValidPerpetualStorageWiggleLocality(perpetualStorageWiggleLocality)) && - LOG_TEST(storageMigrationType != StorageMigrationType::UNSET))) { + LOG_TEST(storageMigrationType != StorageMigrationType::UNSET) && + // sharded-rocksdb maps shards to physical column families using the + // shardId carried only by new-format (shard-encoded) location + // metadata. Old-format metadata cannot describe those mappings, so + // reject configuring original while the storage engine is + // sharded-rocksdb (this also blocks rolling metadata back to + // original without first migrating off sharded-rocksdb). + LOG_TEST(!(shardMetadataFormat == ShardMetadataFormat::ORIGINAL && + storageServerStoreType == KeyValueStoreType::SSD_SHARDED_ROCKSDB)))) { return false; } #undef LOG_TEST @@ -430,6 +440,16 @@ StatusObject DatabaseConfiguration::toJSON(bool noPolicies) const { if (perpetualStoreType.storeType() != KeyValueStoreType::END) { result["perpetual_storage_wiggle_engine"] = perpetualStoreType.toString(); } + if (shardMetadataFormat != ShardMetadataFormat::UNSET) { + result[SHARD_METADATA_FORMAT_KEY] = (shardMetadataFormat == ShardMetadataFormat::ENCODED) + ? SHARD_METADATA_FORMAT_ENCODED + : SHARD_METADATA_FORMAT_ORIGINAL; + } + if (shardMetadataMigration != ShardMetadataMigration::UNSET) { + result[SHARD_METADATA_MIGRATION_KEY] = (shardMetadataMigration == ShardMetadataMigration::ENABLED) + ? SHARD_METADATA_MIGRATION_ENABLED + : SHARD_METADATA_MIGRATION_DISABLED; + } result["storage_migration_type"] = storageMigrationType.toString(); return result; } @@ -454,7 +474,8 @@ std::string DatabaseConfiguration::configureStringFromJSON(const StatusObject& j // For string values, some properties can set with a "=" syntax in "configure" // Such properties are listed here: static std::set directSet = { - "storage_migration_type", "storage_engine", "log_engine", "perpetual_storage_wiggle_engine" + "storage_migration_type", "storage_engine", "log_engine", + "perpetual_storage_wiggle_engine", SHARD_METADATA_FORMAT_KEY, SHARD_METADATA_MIGRATION_KEY }; if (directSet.contains(kv.first)) { @@ -716,6 +737,22 @@ bool DatabaseConfiguration::setInternal(KeyRef key, ValueRef value) { } else if (ck == "perpetual_storage_wiggle_engine"_sr) { parse((&type), value); perpetualStoreType = (KeyValueStoreType::StoreType)type; + } else if (ck == StringRef(SHARD_METADATA_FORMAT_KEY)) { + if (value == StringRef(SHARD_METADATA_FORMAT_ORIGINAL)) { + shardMetadataFormat = ShardMetadataFormat::ORIGINAL; + } else if (value == StringRef(SHARD_METADATA_FORMAT_ENCODED)) { + shardMetadataFormat = ShardMetadataFormat::ENCODED; + } else { + return false; + } + } else if (ck == StringRef(SHARD_METADATA_MIGRATION_KEY)) { + if (value == StringRef(SHARD_METADATA_MIGRATION_DISABLED)) { + shardMetadataMigration = ShardMetadataMigration::DISABLED; + } else if (value == StringRef(SHARD_METADATA_MIGRATION_ENABLED)) { + shardMetadataMigration = ShardMetadataMigration::ENABLED; + } else { + return false; + } } else if (ck == "storage_migration_type"_sr) { parse((&type), value); storageMigrationType = (StorageMigrationType::MigrationType)type; diff --git a/fdbclient/ManagementAPI.cpp b/fdbclient/ManagementAPI.cpp index b160f5765a6..3decea28745 100644 --- a/fdbclient/ManagementAPI.cpp +++ b/fdbclient/ManagementAPI.cpp @@ -185,6 +185,24 @@ std::map configForToken(std::string const& mode) { out[p + key] = format("%d", type); } + if (key == DatabaseConfiguration::SHARD_METADATA_FORMAT_KEY) { + if (value != DatabaseConfiguration::SHARD_METADATA_FORMAT_ORIGINAL && + value != DatabaseConfiguration::SHARD_METADATA_FORMAT_ENCODED) { + printf("Error: shard_metadata_format must be `original' or `encoded'.\n"); + return out; + } + out[p + key] = value; + } + + if (key == DatabaseConfiguration::SHARD_METADATA_MIGRATION_KEY) { + if (value != DatabaseConfiguration::SHARD_METADATA_MIGRATION_ENABLED && + value != DatabaseConfiguration::SHARD_METADATA_MIGRATION_DISABLED) { + printf("Error: shard_metadata_migration must be `enabled' or `disabled'.\n"); + return out; + } + out[p + key] = value; + } + if (key == "exclude") { int p = 0; while (p < value.size()) { diff --git a/fdbclient/Schemas.cpp b/fdbclient/Schemas.cpp index de4d01f85e6..471f8a71ff6 100644 --- a/fdbclient/Schemas.cpp +++ b/fdbclient/Schemas.cpp @@ -891,6 +891,16 @@ const KeyRef JSONSchemas::statusSchema = R"statusSchema( "disabled", "aggressive", "gradual" + ]}, + "shard_metadata_format": { + "$enum":[ + "original", + "encoded" + ]}, + "shard_metadata_migration": { + "$enum":[ + "enabled", + "disabled" ]} }, "consistency_scan" : { diff --git a/fdbclient/SystemData.cpp b/fdbclient/SystemData.cpp index 40d88181e21..471035d094c 100644 --- a/fdbclient/SystemData.cpp +++ b/fdbclient/SystemData.cpp @@ -421,6 +421,14 @@ CheckpointMetaData decodeCheckpointValue(const ValueRef& value) { // "\xff/dataMoves/[[UID]] := [[DataMoveMetaData]]" const KeyRangeRef dataMoveKeys("\xff/dataMoves/"_sr, "\xff/dataMoves0"_sr); + +// See declaration in SystemData.h. Written ("old") and cleared by DD from +// rewriteShardEncodedMetadata; cleared on re-forward by +// clearStaleShardEncodedRewriteSentinel. +const KeyRef shardEncodeMigrationCompleteKey = "\xff/dd/shard_encode_migration_complete"_sr; +const ValueRef shardEncodeMigrationValueOld = "old"_sr; +const ValueRef shardEncodeMigrationValueNew = "new"_sr; + Key dataMoveKeyFor(UID dataMoveId) { BinaryWriter wr(Unversioned()); wr.serializeBytes(dataMoveKeys.begin); diff --git a/fdbclient/include/fdbclient/DatabaseConfiguration.h b/fdbclient/include/fdbclient/DatabaseConfiguration.h index 21d2145e6eb..ee270d709fc 100644 --- a/fdbclient/include/fdbclient/DatabaseConfiguration.h +++ b/fdbclient/include/fdbclient/DatabaseConfiguration.h @@ -222,6 +222,49 @@ struct DatabaseConfiguration { std::string perpetualStorageWiggleLocality; KeyValueStoreType perpetualStoreType; + // Shard-encode metadata configuration (mirrors the storage_engine + + // perpetual_storage_wiggle pair). shard_metadata_format controls how + // DD encodes new shard metadata writes; shard_metadata_migration + // controls whether DD actively converges existing entries to match + // the current format. + // + // Unset -> DD falls back to reading the legacy SHARD_ENCODE_LOCATION_ + // METADATA knob (true -> encoded, false -> original) for the + // format, and treats migration as disabled. Setting shard_metadata_format + // overrides the knob for the encoding target (shard_metadata_migration + // has no knob equivalent; unset simply means "no active rewrite"). + // + // Wire-value constants centralize the strings used by toJSON, + // setInternal, ManagementAPI::configForToken validation, and + // fdbcli's ConfigureCommand help/completion. Keep in sync: a typo + // in one site silently breaks configure round-tripping. + static constexpr const char* SHARD_METADATA_FORMAT_KEY = "shard_metadata_format"; + static constexpr const char* SHARD_METADATA_MIGRATION_KEY = "shard_metadata_migration"; + // "original" = legacy tag-based encoding; "encoded" = shard-encoded + // (UID+dataMoveId) location metadata written under SHARD_ENCODE_LOCATION_METADATA. + static constexpr const char* SHARD_METADATA_FORMAT_ORIGINAL = "original"; + static constexpr const char* SHARD_METADATA_FORMAT_ENCODED = "encoded"; + static constexpr const char* SHARD_METADATA_MIGRATION_ENABLED = "enabled"; + static constexpr const char* SHARD_METADATA_MIGRATION_DISABLED = "disabled"; + enum class ShardMetadataFormat { UNSET, ORIGINAL, ENCODED }; + enum class ShardMetadataMigration { UNSET, DISABLED, ENABLED }; + ShardMetadataFormat shardMetadataFormat; + ShardMetadataMigration shardMetadataMigration; + + // Configured target encoding, or empty when UNSET. Server code applies the + // SHARD_ENCODE_LOCATION_METADATA knob fallback (the knob lives in + // fdbserver, so the fallback cannot be resolved here in fdbclient). + Optional shardMetadataFormatIsEncoded() const { + if (shardMetadataFormat == ShardMetadataFormat::UNSET) { + return Optional(); + } + return shardMetadataFormat == ShardMetadataFormat::ENCODED; + } + // Whether DD should actively converge existing entries at init. + bool shardMetadataMigrationEnabled() const { + return shardMetadataMigration == ShardMetadataMigration::ENABLED; + } + // Storage Migration Type StorageMigrationType storageMigrationType; diff --git a/fdbclient/include/fdbclient/SystemData.h b/fdbclient/include/fdbclient/SystemData.h index de28da09603..f1591ed1c1e 100644 --- a/fdbclient/include/fdbclient/SystemData.h +++ b/fdbclient/include/fdbclient/SystemData.h @@ -176,6 +176,32 @@ Value dataMoveValue(const DataMoveMetaData& dataMove); UID decodeDataMoveKey(const KeyRef& key); DataMoveMetaData decodeDataMoveValue(const ValueRef& value); +// Migration-state sentinel: written by DD after a shard-encode +// migration rewrite pass completes with no more work in the current +// direction. Fast-path skip for subsequent DD inits — a single-key +// read tells DD whether the metadata is already at the target +// encoding for the effective shard_metadata_format target (config, or +// the SHARD_ENCODE_LOCATION_METADATA knob when the config is UNSET). +// +// Values: +// "old" -> rollback (tag-based) rewrite drained; safe to downgrade. +// "new" -> reserved for a sealed forward completion; NOT currently +// written (see below). +// absent -> unknown / in-progress. In the rollback direction DD does +// the full scan. In the forward direction DD does not scan; +// it only clears a stale "old" sentinel if present. +// +// The rollback rewriter CLEARS this key as the first commit of its pass +// (so an audit tool observing the cluster during a rewrite doesn't see a +// stale "complete" marker) and SETS it to "old" only when a pass finds no +// more new-format entries anywhere. The forward path does NOT seal a +// "new" value — it only clears any stale "old" +// (clearStaleShardEncodedRewriteSentinel); forward completion is reached +// lazily by natural DD moves and is not marked here. +extern const KeyRef shardEncodeMigrationCompleteKey; +extern const ValueRef shardEncodeMigrationValueOld; +extern const ValueRef shardEncodeMigrationValueNew; + // "\xff/serverKeys/[[serverID]]/[[begin]]" := "[[serverKeysTrue]]" |" [[serverKeysFalse]]" // An internal mapping of what shards any given server currently has ownership of // Using the serverID as a prefix, then followed by the beginning of the shard range diff --git a/fdbserver/clustercontroller/ClusterRecovery.cpp b/fdbserver/clustercontroller/ClusterRecovery.cpp index fc30414f4b7..e0bdae2335e 100644 --- a/fdbserver/clustercontroller/ClusterRecovery.cpp +++ b/fdbserver/clustercontroller/ClusterRecovery.cpp @@ -1898,7 +1898,11 @@ Future clusterRecoveryCore(Reference self) { } else { // Recruit and seed initial shard servers // This transaction must be the very first one in the database (version 1) - seedShardServers(recoveryCommitRequest.arena, tr, seedServers); + seedShardServers(recoveryCommitRequest.arena, + tr, + seedServers, + self->configuration.shardMetadataFormatIsEncoded().orDefault( + SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA)); } // initialConfChanges have not been conflict checked against any earlier writes in the recovery transaction, so do // this as early as possible in the recovery transaction but see above comments as to why it can't be absolutely diff --git a/fdbserver/core/MoveKeys.cpp b/fdbserver/core/MoveKeys.cpp index 3367154f286..c781aae90bc 100644 --- a/fdbserver/core/MoveKeys.cpp +++ b/fdbserver/core/MoveKeys.cpp @@ -229,6 +229,8 @@ Future deleteCheckpoints(Transaction* tr, std::set checkpointIds, UID } } // namespace +DDEnabledState::DDEnabledState() : shardMetadataFormatIsNew_(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) {} + bool DDEnabledState::sameId(const UID& id) const { return ddEnabledStatusUID == id; } @@ -236,6 +238,14 @@ bool DDEnabledState::isEnabled() const { return stateValue == ENABLED; } +bool DDEnabledState::shardEncodeLocationMetadata() const { + return shardMetadataFormatIsNew_; +} + +void DDEnabledState::setShardEncodeLocationMetadata(bool isNewFormat) { + shardMetadataFormatIsNew_ = isNewFormat; +} + bool DDEnabledState::isBlobRestorePreparing() const { return stateValue == BLOB_RESTORE_PREPARING; } @@ -512,11 +522,12 @@ Future auditLocationMetadataPreCheck(Database occ, KeyRange range, std::vector servers, std::string context, - UID dataMoveId) { - if (!SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + UID dataMoveId, + const DDEnabledState* ddEnabledState) { + if (!ddEnabledState->shardEncodeLocationMetadata()) { throw dd_config_changed(); } - ASSERT(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + ASSERT(ddEnabledState->shardEncodeLocationMetadata()); if (range.empty()) { TraceEvent(SevWarn, "CheckLocationMetadataEmptyInputRange").detail("By", "PreCheck").detail("Range", range); @@ -582,11 +593,15 @@ Future auditLocationMetadataPreCheck(Database occ, } } -Future auditLocationMetadataPostCheck(Database occ, KeyRange range, std::string context, UID dataMoveId) { - if (!SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { +Future auditLocationMetadataPostCheck(Database occ, + KeyRange range, + std::string context, + UID dataMoveId, + const DDEnabledState* ddEnabledState) { + if (!ddEnabledState->shardEncodeLocationMetadata()) { throw dd_config_changed(); } - ASSERT(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + ASSERT(ddEnabledState->shardEncodeLocationMetadata()); if (range.empty()) { TraceEvent(g_network->isSimulated() ? SevError : SevWarnAlways, "CheckLocationMetadataEmptyInputRange") @@ -720,10 +735,10 @@ Future cleanUpSingleShardDataMove(Database occ, FlowLock* cleanUpDataMoveParallelismLock, UID dataMoveId, const DDEnabledState* ddEnabledState) { - if (!SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (!ddEnabledState->shardEncodeLocationMetadata()) { throw dd_config_changed(); } - ASSERT(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + ASSERT(ddEnabledState->shardEncodeLocationMetadata()); TraceEvent(SevInfo, "CleanUpSingleShardDataMoveBegin", dataMoveId).detail("Range", keys); static auto* counters = makeCounters("/movekeys/cleanUpSingleShardDataMove"); @@ -751,7 +766,7 @@ Future cleanUpSingleShardDataMove(Database occ, throw operation_cancelled(); } if (currentShards.empty()) { - if (!SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (!ddEnabledState->shardEncodeLocationMetadata()) { throw dd_config_changed(); } ASSERT(!currentShards.empty()); @@ -779,7 +794,7 @@ Future cleanUpSingleShardDataMove(Database occ, std::vector servers(src.size() + dest.size()); std::merge(src.begin(), src.end(), dest.begin(), dest.end(), servers.begin()); co_await auditLocationMetadataPreCheck( - occ, &tr, keys, servers, "cleanUpSingleShardDataMove_precheck", dataMoveId); + occ, &tr, keys, servers, "cleanUpSingleShardDataMove_precheck", dataMoveId, ddEnabledState); } TraceEvent(SevInfo, "CleanUpSingleShardDataMove", dataMoveId) @@ -808,7 +823,8 @@ Future cleanUpSingleShardDataMove(Database occ, // Post validate consistency of update of keyServers and serverKeys if (SERVER_KNOBS->AUDIT_DATAMOVE_POST_CHECK) { - co_await auditLocationMetadataPostCheck(occ, keys, "cleanUpSingleShardDataMove_postcheck", dataMoveId); + co_await auditLocationMetadataPostCheck( + occ, keys, "cleanUpSingleShardDataMove_postcheck", dataMoveId, ddEnabledState); } break; } catch (Error& e) { @@ -2165,7 +2181,7 @@ static Future startMoveShards(Database occ, std::vector servers(src.size() + dest.size()); std::merge(src.begin(), src.end(), dest.begin(), dest.end(), servers.begin()); co_await auditLocationMetadataPreCheck( - occ, &tr, rangeIntersectKeys, servers, "startMoveShards_precheck", dataMoveId); + occ, &tr, rangeIntersectKeys, servers, "startMoveShards_precheck", dataMoveId, ddEnabledState); } if (destId.isValid()) { @@ -2335,7 +2351,8 @@ static Future startMoveShards(Database occ, if (currentKeys.end == keys.end) { // Post validate consistency of update of keyServers and serverKeys if (SERVER_KNOBS->AUDIT_DATAMOVE_POST_CHECK) { - co_await auditLocationMetadataPostCheck(occ, keys, "startMoveShards_postcheck", dataMoveId); + co_await auditLocationMetadataPostCheck( + occ, keys, "startMoveShards_postcheck", dataMoveId, ddEnabledState); } break; } @@ -2455,8 +2472,9 @@ static Future decodeAndPreCheckShards(Database occ, bool runPreCheck, DataMoveMetaData const& dataMove, UID relocationIntervalId, - Severity sevDm, - bool* cancelDataMove) { + Severity sevDm, + bool* cancelDataMove, + const DDEnabledState* ddEnabledState) { std::vector completeSrc; std::unordered_set allServers; @@ -2490,7 +2508,7 @@ static Future decodeAndPreCheckShards(Database occ, std::vector servers(src.size() + dest.size()); std::merge(src.begin(), src.end(), dest.begin(), dest.end(), servers.begin()); co_await auditLocationMetadataPreCheck( - occ, tr, currentRange, servers, "finishMoveShards_precheck", dataMoveId); + occ, tr, currentRange, servers, "finishMoveShards_precheck", dataMoveId, ddEnabledState); } std::sort(dest.begin(), dest.end()); @@ -2948,7 +2966,8 @@ static Future finishMoveShards(Database occ, dataMove, relocationIntervalId, sevDm, - &cancelDataMove); + &cancelDataMove, + ddEnabledState); // Read the destination SSes' interfaces and record the read // version waitForShardReady will need after we drop the txn. @@ -3030,7 +3049,8 @@ static Future finishMoveShards(Database occ, co_await auditLocationMetadataPostCheck(occ, postWaitDataMove.ranges.front(), "finishMoveShards_postcheck", - relocationIntervalId); + relocationIntervalId, + ddEnabledState); } break; } @@ -3568,7 +3588,7 @@ Future removeKeysFromFailedServer(Database cx, DataMovementReason::ASSIGN_EMPTY_RANGE); // Assign the shard to teamForDroppedRange in keyServer space. - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { tr.set(keyServersKey(it.key), keyServersValue(teamForDroppedRange, {}, shardId, UID())); } else { tr.set(keyServersKey(it.key), keyServersValue(UIDtoTagMap, teamForDroppedRange)); @@ -3587,7 +3607,7 @@ Future removeKeysFromFailedServer(Database cx, // Note, there could be data loss. std::vector> emptyRangeActors; for (const UID& id : teamForDroppedRange) { - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { emptyRangeActors.push_back(krmSetRangeCoalescing( &tr, serverKeysPrefixFor(id), range, allKeys, serverKeysValue(shardId))); } else { @@ -3610,7 +3630,13 @@ Future removeKeysFromFailedServer(Database cx, .detail("Key", it.key) .detail("ValueSrc", describe(src)) .detail("ValueDest", describe(dest)); - if (srcId != anonymousShardId) { + // Only preserve new-format (shardId-encoded) keyServers when the + // cluster is writing new format. During a rollback (target + // original) this failed-server cleanup must not re-introduce a + // new-format entry behind the DD-init rewrite — write old + // (tag-based) format. Mirrors the gated drop-all-replicas branch + // above. + if (srcId != anonymousShardId && ddEnabledState->shardEncodeLocationMetadata()) { if (dest.empty()) destId = UID(); tr.set(keyServersKey(it.key), keyServersValue(src, dest, srcId, destId)); @@ -3805,7 +3831,7 @@ Future cleanUpDataMoveCore(Database occ, std::vector servers(src.size() + dest.size()); std::merge(src.begin(), src.end(), dest.begin(), dest.end(), servers.begin()); co_await auditLocationMetadataPreCheck( - occ, &tr, rangeIntersectKeys, servers, "cleanUpDataMoveCore_precheck", dataMoveId); + occ, &tr, rangeIntersectKeys, servers, "cleanUpDataMoveCore_precheck", dataMoveId, ddEnabledState); } for (const auto& uid : src) { @@ -3840,10 +3866,17 @@ Future cleanUpDataMoveCore(Database occ, oldDests.insert(uid); } + // During a rollback (target original) this physical-datamove + // cleanup must not re-introduce a new-format (shardId-encoded) + // keyServers entry behind the DD-init rewrite — write old + // (tag-based) format. + Value cleanupKsValue = ddEnabledState->shardEncodeLocationMetadata() + ? keyServersValue(src, {}, srcId, UID()) + : keyServersValue(UIDtoTagMap, src, {}); krmSetPreviouslyEmptyRange(&tr, keyServersPrefix, rangeIntersectKeys, - keyServersValue(src, {}, srcId, UID()), + cleanupKsValue, currentShards[i + 1].value); } @@ -3880,7 +3913,7 @@ Future cleanUpDataMoveCore(Database occ, // Post validate consistency of update of keyServers and serverKeys if (SERVER_KNOBS->AUDIT_DATAMOVE_POST_CHECK) { co_await auditLocationMetadataPostCheck( - occ, dataMove.ranges.front(), "cleanUpDataMoveCore_postcheck", dataMoveId); + occ, dataMove.ranges.front(), "cleanUpDataMoveCore_postcheck", dataMoveId, ddEnabledState); } break; } @@ -3943,9 +3976,10 @@ Future cleanUpDataMove(Database occ, Future rawStartMovement(Database occ, const MoveKeysParams& params, std::map& tssMapping) { - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { - // A relocation launched before the knob changed can still carry the old-path sentinel. - // Never persist it as a shard-encoded data move; restart DD with the new configuration. + if (params.ddEnabledState->shardEncodeLocationMetadata()) { + // A relocation launched before the target format changed can still carry the old-path + // sentinel. Never persist it as a shard-encoded data move; restart DD with the new + // configuration. if (!params.ranges.present() || params.dataMoveId == anonymousShardId) { throw dd_config_changed(); } @@ -3978,7 +4012,7 @@ Future rawStartMovement(Database occ, Future rawCheckFetchingState(const Database& cx, const MoveKeysParams& params, const std::map& tssMapping) { - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (params.ddEnabledState->shardEncodeLocationMetadata()) { if (!params.ranges.present()) { throw dd_config_changed(); } @@ -4006,7 +4040,7 @@ Future rawCheckFetchingState(const Database& cx, Future rawFinishMovement(Database occ, const MoveKeysParams& params, const std::map& tssMapping) { - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (params.ddEnabledState->shardEncodeLocationMetadata()) { if (!params.ranges.present()) { throw dd_config_changed(); } diff --git a/fdbserver/core/SeedShardServers.cpp b/fdbserver/core/SeedShardServers.cpp index 4bb304bc27f..b932ce7fc40 100644 --- a/fdbserver/core/SeedShardServers.cpp +++ b/fdbserver/core/SeedShardServers.cpp @@ -29,7 +29,10 @@ #include "fdbserver/core/SeedShardServers.h" #include "flow/Trace.h" -void seedShardServers(Arena& arena, CommitTransactionRef& tr, std::vector servers) { +void seedShardServers(Arena& arena, + CommitTransactionRef& tr, + std::vector servers, + bool shardEncodeLocationMetadata) { std::map, Tag> dcIdLocality; std::map serverTag; int8_t nextLocality = 0; @@ -75,7 +78,7 @@ void seedShardServers(Arena& arena, CommitTransactionRef& tr, std::vectorTAG_ENCODE_KEY_SERVERS ? keyServersValue(serverTags) : keyServersValue(RangeResult(), serverSrcIds); - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (shardEncodeLocationMetadata) { const UID shardId = newDataMoveId(deterministicRandom()->randomUInt64(), AssignEmptyRange(false), DataMoveType::LOGICAL, diff --git a/fdbserver/core/ServerKnobs.cpp b/fdbserver/core/ServerKnobs.cpp index d37d8d6ab9c..795ccb44a5c 100644 --- a/fdbserver/core/ServerKnobs.cpp +++ b/fdbserver/core/ServerKnobs.cpp @@ -332,6 +332,7 @@ void ServerKnobs::initialize(Randomize randomize, ClientKnobs* clientKnobs, IsSi init( ALLOW_LARGE_SHARD, false ); if( randomize && buggify() ) ALLOW_LARGE_SHARD = true; init( MAX_LARGE_SHARD_BYTES, 1000000000 ); // 1G init( SHARD_ENCODE_LOCATION_METADATA, false ); if( isSimulated ) { bool v = deterministicRandom()->random01() < 0.75; if( !explicitlySetKnobs.contains("shard_encode_location_metadata") ) SHARD_ENCODE_LOCATION_METADATA = v; } + init( SHARD_ENCODE_REWRITE_KS_BATCH_SIZE, 1000 ); if( randomize && buggify() ) SHARD_ENCODE_REWRITE_KS_BATCH_SIZE = deterministicRandom()->randomInt(1, 11); // small batch forces multi-page Phase 2 pagination in sim init( ENABLE_DD_PHYSICAL_SHARD, false ); // EXPERIMENTAL; If true, SHARD_ENCODE_LOCATION_METADATA must be true; When true, optimization of data move between DCs is disabled init( DD_PHYSICAL_SHARD_MOVE_PROBABILITY, 0.0 ); // FIXME: re-enable after ShardedRocksDB is well tested by simulation init( ENABLE_PHYSICAL_SHARD_MOVE_EXPERIMENT, false ); // FIXME: re-enable after ShardedRocksDB is well tested by simulation diff --git a/fdbserver/core/include/fdbserver/core/Knobs.h b/fdbserver/core/include/fdbserver/core/Knobs.h index ca1c827cc73..9209c1585ad 100644 --- a/fdbserver/core/include/fdbserver/core/Knobs.h +++ b/fdbserver/core/include/fdbserver/core/Knobs.h @@ -252,7 +252,17 @@ class SWIFT_CXX_IMMORTAL_SINGLETON_TYPE ServerKnobs : public KnobsImpl servers); +void seedShardServers(Arena& arena, + CommitTransactionRef& tr, + std::vector servers, + bool shardEncodeLocationMetadata); #endif diff --git a/fdbserver/datadistributor/DDRelocationQueue.cpp b/fdbserver/datadistributor/DDRelocationQueue.cpp index 764703b4c04..84a679565f8 100644 --- a/fdbserver/datadistributor/DDRelocationQueue.cpp +++ b/fdbserver/datadistributor/DDRelocationQueue.cpp @@ -1255,7 +1255,7 @@ void DDQueue::launchQueuedWork(std::set // which leads to conflicts of moving keys Future fCleanup = - SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA ? cancelDataMove(this, rd.keys, ddEnabledState) : Void(); + ddEnabledState->shardEncodeLocationMetadata() ? cancelDataMove(this, rd.keys, ddEnabledState) : Void(); inFlight.insert(rd.keys, rd); for (int r = 0; r < ranges.size(); r++) { @@ -1271,14 +1271,14 @@ void DDQueue::launchQueuedWork(std::set } if (rd.keys == ranges[r] && rd.isRestore()) { ASSERT(rd.dataMove != nullptr); - ASSERT(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + ASSERT(ddEnabledState->shardEncodeLocationMetadata()); rrs.dataMoveId = rd.dataMove->meta.id; } else { // Restored data moves can split ordinary relocations, but not other restored data moves. ASSERT_WE_THINK(!rd.isRestore() || !rrs.isRestore()); // TODO(psm): The shard id is determined by DD. rrs.dataMove.reset(); - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { if (SERVER_KNOBS->ENABLE_DD_PHYSICAL_SHARD) { rrs.dataMoveId = UID(); } else if (rrs.bulkLoadTask.present()) { @@ -1610,7 +1610,7 @@ Future dataDistributionRelocator(DDQueue* self, self->suppressIntervals = 0; } - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { auto inFlightRange = self->inFlight.rangeContaining(rd.keys.begin); ASSERT(inFlightRange.range() == rd.keys); ASSERT(inFlightRange.value().randomId == rd.randomId); @@ -1740,7 +1740,7 @@ Future dataDistributionRelocator(DDQueue* self, bestTeams.clear(); // Get team from teamCollections in different DCs and find the best one while (tciIndex < self->teamCollections.size()) { - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA && rd.isRestore()) { + if (ddEnabledState->shardEncodeLocationMetadata() && rd.isRestore()) { auto req = GetTeamRequest(tciIndex == 0 ? rd.dataMove->primaryDest : rd.dataMove->remoteDest); req.keys = rd.keys; Future>, bool>> fbestTeam = @@ -2203,7 +2203,7 @@ Future dataDistributionRelocator(DDQueue* self, Promise dataMovementComplete; // Move keys from source to destination by changing the serverKeyList and keyServerList system keys std::unique_ptr params; - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { params = std::make_unique(rd.dataMoveId, std::vector{ rd.keys }, destIds, @@ -2235,7 +2235,7 @@ Future dataDistributionRelocator(DDQueue* self, : Optional()); } Future doMoveKeys = self->txnProcessor->moveKeys(*params); - Future pollHealth = signalledTransferComplete && !SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA + Future pollHealth = signalledTransferComplete && !ddEnabledState->shardEncodeLocationMetadata() ? Never() : delay(SERVER_KNOBS->HEALTH_POLL_TIME, TaskPriority::DataDistributionLaunch); try { @@ -2250,7 +2250,7 @@ Future dataDistributionRelocator(DDQueue* self, extraIds.clear(); ASSERT(totalIds == destIds.size()); // Sanity check the destIDs before we move keys std::unique_ptr params; - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { params = std::make_unique(rd.dataMoveId, std::vector{ rd.keys }, destIds, @@ -2286,7 +2286,7 @@ Future dataDistributionRelocator(DDQueue* self, doMoveKeys = self->txnProcessor->moveKeys(*params); } else { self->fetchKeysComplete.insert(rd); - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { auto ranges = self->dataMoves.getAffectedRangesAfterInsertion(rd.keys); if (ranges.size() == 1 && static_cast(ranges[0]) == rd.keys && ranges[0].value.id == rd.dataMoveId && !ranges[0].value.cancel.isValid()) { @@ -2305,7 +2305,7 @@ Future dataDistributionRelocator(DDQueue* self, self->dataTransferComplete.send(rd); ownsDestBusyness = false; } - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + if (ddEnabledState->shardEncodeLocationMetadata()) { CODE_PROBE(true, "Cancel shard-encoded data move after destination team becomes unhealthy"); TraceEvent(SevWarnAlways, "RelocateShardDestinationTeamUnhealthy", distributorId) @@ -2320,7 +2320,7 @@ Future dataDistributionRelocator(DDQueue* self, throw data_move_dest_team_not_found(); } } - pollHealth = signalledTransferComplete && !SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA + pollHealth = signalledTransferComplete && !ddEnabledState->shardEncodeLocationMetadata() ? Never() : delay(SERVER_KNOBS->HEALTH_POLL_TIME, TaskPriority::DataDistributionLaunch); } else if (res.index() == 2) { diff --git a/fdbserver/datadistributor/DDShardTracker.cpp b/fdbserver/datadistributor/DDShardTracker.cpp index e74a513c00c..74dee3cf9f3 100644 --- a/fdbserver/datadistributor/DDShardTracker.cpp +++ b/fdbserver/datadistributor/DDShardTracker.cpp @@ -152,7 +152,7 @@ std::pair calculateShardSizeBounds( } Future shardUsableRegions(DataDistributionTracker::SafeAccessor self, KeyRange keys) { - ASSERT(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + ASSERT(self()->shardEncodeLocationMetadata); ASSERT(SERVER_KNOBS->DD_SHARD_USABLE_REGION_CHECK_RATE > 0); co_await yieldedFuture(self()->readyToStart.getFuture()); double expectedCompletionSeconds = self()->shards->size() * 1.0 / SERVER_KNOBS->DD_SHARD_USABLE_REGION_CHECK_RATE; @@ -975,7 +975,7 @@ void restartShardTrackers(DataDistributionTracker* self, data.trackShard = shardTracker(DataDistributionTracker::SafeAccessor(self), ranges[i], shardMetrics); data.trackBytes = trackShardMetrics(DataDistributionTracker::SafeAccessor(self), ranges[i], shardMetrics, whenDDInit); - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA && SERVER_KNOBS->DD_SHARD_USABLE_REGION_CHECK_RATE > 0 && + if (self->shardEncodeLocationMetadata && SERVER_KNOBS->DD_SHARD_USABLE_REGION_CHECK_RATE > 0 && self->usableRegions != -1) { data.trackUsableRegion = shardUsableRegions(DataDistributionTracker::SafeAccessor(self), ranges[i]); } @@ -1237,7 +1237,8 @@ DataDistributionTracker::DataDistributionTracker(DataDistributionTrackerInitPara output(params.output), shardsAffectedByTeamFailure(params.shardsAffectedByTeamFailure), physicalShardCollection(params.physicalShardCollection), bulkLoadTaskCollection(params.bulkLoadTaskCollection), readyToStart(params.readyToStart), anyZeroHealthyTeams(params.anyZeroHealthyTeams), - trackerCancelled(params.trackerCancelled), usableRegions(params.usableRegions) {} + trackerCancelled(params.trackerCancelled), usableRegions(params.usableRegions), + shardEncodeLocationMetadata(params.shardEncodeLocationMetadata) {} DataDistributionTracker::~DataDistributionTracker() { if (trackerCancelled) { @@ -1365,7 +1366,7 @@ Future DataDistributionTracker::run( self->triggerStorageQueueRebalance = triggerStorageQueueRebalance; self->triggerShardBulkLoading = triggerShardBulkLoading; self->userRangeConfig = initData->userRangeConfig; - self->bulkLoadEnabled = bulkLoadIsEnabled(initData->bulkLoadMode); + self->bulkLoadEnabled = bulkLoadIsEnabled(initData->bulkLoadMode, self->shardEncodeLocationMetadata); return holdWhile(self, DataDistributionTrackerImpl::run(self.getPtr(), initData)); } diff --git a/fdbserver/datadistributor/DDShardTracker.h b/fdbserver/datadistributor/DDShardTracker.h index 2a88d3c6071..c26625288c2 100644 --- a/fdbserver/datadistributor/DDShardTracker.h +++ b/fdbserver/datadistributor/DDShardTracker.h @@ -49,6 +49,9 @@ struct DataDistributionTrackerInitParams { KeyRangeMap* shards = nullptr; bool* trackerCancelled = nullptr; int32_t usableRegions = -1; + // Effective shard-location-metadata encoding target (resolved from + // DatabaseConfiguration with knob fallback, published on ddEnabledState). + bool shardEncodeLocationMetadata = false; }; // track the status of shards @@ -80,6 +83,10 @@ class DataDistributionTracker : public IDDShardTracker, public ReferenceCounted< Reference bulkLoadTaskCollection; bool bulkLoadEnabled = false; + // Effective shard-location-metadata encoding target for this DD + // incarnation (mirrors DDEnabledState::shardEncodeLocationMetadata()). + bool shardEncodeLocationMetadata = false; + Promise readyToStart; Reference> anyZeroHealthyTeams; diff --git a/fdbserver/datadistributor/DDTeamCollection.actor.cpp b/fdbserver/datadistributor/DDTeamCollection.actor.cpp index 8104a8dace2..b680cde6f21 100644 --- a/fdbserver/datadistributor/DDTeamCollection.actor.cpp +++ b/fdbserver/datadistributor/DDTeamCollection.actor.cpp @@ -3500,7 +3500,8 @@ class DDTeamCollectionImpl { // Update server's storeType, especially when it was created wait(server->updateStoreType()); if (server->getStoreType() == KeyValueStoreType::SSD_SHARDED_ROCKSDB && - !SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + !self->configuration.shardMetadataFormatIsEncoded().orDefault( + SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA)) { TraceEvent(SevError, "PhysicalShardNotEnabledForShardedRocks", self->getDistributorId()) .detail("StorageServer", server->getId()); throw internal_error(); diff --git a/fdbserver/datadistributor/DDTxnProcessor.cpp b/fdbserver/datadistributor/DDTxnProcessor.cpp index 2ae32a7612b..097cd065c55 100644 --- a/fdbserver/datadistributor/DDTxnProcessor.cpp +++ b/fdbserver/datadistributor/DDTxnProcessor.cpp @@ -21,6 +21,7 @@ #include "DDTxnProcessor.h" #include "fdbclient/NativeAPI.actor.h" #include "fdbclient/ManagementAPI.h" +#include "fdbclient/RunRYWTransaction.h" #include "DataDistribution.h" #include "fdbclient/DatabaseContext.h" #include "flow/TxnCounters.h" @@ -300,62 +301,57 @@ class DDTxnProcessorImpl { co_return Optional(); } - // When SHARD_ENCODE_LOCATION_METADATA is false on DD init, do a bounded - // rewrite to start the rollback. Two phases, both bounded: + // Rollback rewrite invoked at DD init when the migration target is + // old format: converts shard-encoded metadata back to old format so + // `audit_storage metadata_encoding` can reach ROLLBACK COMPLETE. + // See design/shard-encode-location-metadata.md for the full design. // - // Phase 1: Clear all DataMoveMetaData (single transaction; the dataMoves - // keyspace is small, this is always one commit). + // Three phases: (1) clear DataMoveMetaData, (2) rewrite keyServers + // (paginated, re-entrant), (3) rewrite each SS's serverKeys KRM. A + // completion sentinel lets later DD inits fast-path skip. // - // Phase 2: Rewrite up to 1000 keyServers entries at the head of the - // prefix from new (UID-based) to old (tag-based) format. Returns true - // if either phase committed; the caller restarts the outer init loop - // and calls back in. The Phase-2 cap means clusters with more than - // 1000 shard-encoded keyServers entries are NOT fully rewritten by - // this function — by design. Bulk rewrite happens through normal - // shard movement / storage wiggle once the knob is false (see the - // "Migration for downgrade" section of - // design/shard-encode-location-metadata.md); this function just - // clears dataMoves and rewrites the small remnant at the head so DD - // init has a tidy starting point. - // - // Calling this when nothing needs rewriting (knob has been false the - // whole time, or rollback already complete) is safe and - // write-cost-free: the reads find no shard-encoded entries, no - // commits happen, returns false. Cost is three system-key reads on - // every DD init when knob is false. - // - // serverKeys entries are left in place — they drain naturally as DD - // moves shards using the old path. - static Future rewriteShardEncodedMetadata(Transaction& tr, UID distributorId) { - TraceEvent(SevInfo, "DDInitShardEncodeOff", distributorId) - .detail("KnobValue", SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); - tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); - tr.setOption(FDBTransactionOptions::LOCK_AWARE); - - // Phase 1: Clear all DataMoveMetaData + // Two non-obvious properties: + // - The Phase 3 serverKeys rewrite is required for convergence; + // without it, new-format serverKeys only drain opportunistically + // as DD moves shards, so ROLLBACK COMPLETE may never be reached. + // - Phase 3 is NOT re-entrant: it drains all SSes in one DD-init + // invocation (up to ~3.5h serial on a 250k-shard cluster) before + // setting the sentinel. Rare, opt-in, runs per rollback event + // (configure shard_metadata_migration=enabled). + // Phase 1 of rewriteShardEncodedMetadata: clear all persisted + // DataMoveMetaData entries. New-format dataMoves are meaningless to + // old-format DD; on rollback DD needs a clean slate to plan moves. + // Any clear commits on the caller's tr (batched with the sentinel- + // clear that the top-level function queued on the same tr). Returns + // true if any were cleared — caller re-enters until this returns + // false. + static Future clearShardEncodedDataMoves(Transaction& tr, UID distributorId) { RangeResult dmsCheck = co_await tr.getRange(dataMoveKeys, CLIENT_KNOBS->TOO_MANY); ASSERT(!dmsCheck.more && dmsCheck.size() < CLIENT_KNOBS->TOO_MANY); - if (!dmsCheck.empty()) { - TraceEvent(SevWarnAlways, "DDInitCancellingShardEncodedMoves", distributorId) - .detail("Count", dmsCheck.size()); - tr.clear(dataMoveKeys); - co_await tr.commit(); - tr.reset(); - tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); - tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); - tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); - co_return true; + if (dmsCheck.empty()) { + co_return false; } + TraceEvent(SevWarnAlways, "DDInitCancellingShardEncodedMoves", distributorId).detail("Count", dmsCheck.size()); + tr.clear(dataMoveKeys); + co_await tr.commit(); + tr.reset(); + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + co_return true; + } - // Phase 2: Rewrite shard-encoded keyServers entries to old format. - // Reads 1000 entries per iteration. Caller loops (co_return true triggers - // re-entry) until no shard-encoded entries remain. Previously-rewritten - // entries won't match hasShardEncodeLocationMetaData() on re-read. + // Phase 2: rewrite one batch (SHARD_ENCODE_REWRITE_KS_BATCH_SIZE + // entries) of shard-encoded keyServers entries to old format. Caller + // owns `beginKey` and re-enters until this returns false, so the + // cursor persists across calls to cover the whole prefix. + static Future rewriteShardEncodedKeyServers(Transaction& tr, UID distributorId, Key& beginKey) { RangeResult UIDtoTagMap = co_await tr.getRange(serverTagKeys, CLIENT_KNOBS->TOO_MANY); ASSERT(!UIDtoTagMap.more && UIDtoTagMap.size() < CLIENT_KNOBS->TOO_MANY); bool rewroteAny = false; - RangeResult ksEntries = co_await tr.getRange(KeyRangeRef(keyServersPrefix, keyServersEnd), 1000); + RangeResult ksEntries = co_await tr.getRange(KeyRangeRef(beginKey, keyServersEnd), + SERVER_KNOBS->SHARD_ENCODE_REWRITE_KS_BATCH_SIZE); for (const auto& kv : ksEntries) { if (kv.value.empty()) continue; @@ -369,19 +365,408 @@ class DDTxnProcessorImpl { rewroteAny = true; } } - - if (rewroteAny) { - TraceEvent(SevInfo, "DDInitRewritingShardEncodedMetadata", distributorId) - .detail("KeyServersEntries", ksEntries.size()); - co_await tr.commit(); - tr.reset(); + // Compute the next cursor, but DO NOT publish it to the caller-owned + // `beginKey` until AFTER a successful commit below. This tr carries + // the caller's DD-init read set (serverList/workers/mode keys), so + // its commit can throw not_committed on a conflict — common under + // churn (e.g. Attrition changing the server list). If we advanced + // beginKey before committing, the caller's retry would resume from + // the already-advanced cursor and permanently skip this batch's + // unconverted new-format entries. That is the root cause of the + // KeyServersNew residual / early-seal (seeds 2611177188, 1561219216): + // only the FIRST batch leaked, because after tr.reset() below later + // batches carry a small read set and rarely conflict. + const bool atEnd = !ksEntries.more; + Key nextBegin = ksEntries.empty() ? beginKey : keyAfter(ksEntries.back().key); + if (atEnd) { + nextBegin = keyServersPrefix; // defensive: caller won't re-enter after false + } + if (!rewroteAny) { + // Only safe to declare Phase 2 done once the whole prefix is + // walked. If more pages remain we must re-enter — returning + // false here would reach ROLLBACK COMPLETE while unrewritten + // entries remain further along the prefix. + if (atEnd) { + co_return false; // done; leave tr for the caller's sentinel-clear commit + } + beginKey = nextBegin; // read-only page (no commit to fail): safe to advance + tr.reset(); // no writes this page; drop read set before re-entering tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); co_return true; } + TraceEvent(SevInfo, "DDInitRewritingShardEncodedMetadata", distributorId) + .detail("KeyServersEntries", ksEntries.size()) + .detail("More", ksEntries.more); + co_await tr.commit(); + beginKey = nextBegin; // publish the cursor ONLY after the durable commit + tr.reset(); + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + co_return true; + } + + // Phase 3 per-SS: rewrite one SS's serverKeys KRM from new-format + // entries to old-format constants. Uses ONE FRESH TRANSACTION PER + // krmSetRangeCoalescing CALL — same pattern natural DD moves use in + // MoveKeys.cpp:99. See rewriteShardEncodedMetadata's Phase 3 comment + // for the correctness argument. Returns the number of ranges + // rewritten across the entire SS (paginates internally so callers + // don't need to re-invoke to finish one SS). + static Future rewriteOneServerKeysKRM(Database cx, UID ssId, UID distributorId, bool& fullyDrained) { + Key mapPrefix = serverKeysPrefixFor(ssId); + int64_t rewritesForThisSS = 0; + double ssStart = now(); + int scanIdx = 0; + // True unless we bail below with possible residue (no-progress or + // scan-limit break). The caller must NOT seal the completion + // sentinel if any SS returns fullyDrained=false, else it would + // report ROLLBACK COMPLETE while new-format serverKeys remain. + fullyDrained = true; + + // Repeat full KRM scans until a complete scan finds no new-format + // spans. A single forward paginated pass can leave residue (a + // no-progress break under a small KRM_GET_RANGE_LIMIT, a partial + // page, or coalescing shifting boundaries mid-pass), so we must not + // return "drained" until a whole scan rewrote nothing. + loop { + scanIdx++; + int64_t rewritesThisScan = 0; + bool noProgress = false; + Key beginKey = allKeys.begin; + int pageIdx = 0; + + while (beginKey < allKeys.end) { + // Read one page of this SS's KRM in its own fresh snapshot + // transaction. KRM_GET_RANGE_LIMIT bounds page size; under + // buggify this can be as small as 10 entries. + RangeResult ranges; + bool morePages = false; + { + Transaction readTr(cx); + readTr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + readTr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + readTr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + + // Guard: krmGetRanges hangs on an empty KRM prefix + // (freshly registered SS with no assignments yet). + if (beginKey == allKeys.begin) { + RangeResult probe = co_await readTr.getRange(KeyRangeRef(mapPrefix, strinc(mapPrefix)), 1); + if (probe.empty()) { + co_return rewritesForThisSS; + } + } + + ranges = co_await krmGetRanges(&readTr, + mapPrefix, + KeyRangeRef(beginKey, allKeys.end), + CLIENT_KNOBS->KRM_GET_RANGE_LIMIT, + CLIENT_KNOBS->KRM_GET_RANGE_LIMIT_BYTES); + morePages = ranges.more; + } + + if (scanIdx == 1 && pageIdx == 0) { + TraceEvent(SevInfo, "DDShardEncodeRollbackPhase3SSStart", distributorId) + .detail("SS", ssId) + .detail("FirstPageEntries", ranges.size()); + } + + for (int i = 0; i + 1 < ranges.size(); i++) { + const ValueRef& v = ranges[i].value; + if (isServerKeysUnassigned(v) || isServerKeysOldFormatAssigned(v)) { + continue; + } + bool assigned = false; + bool emptyRange = false; + DataMoveType dataMoveType = DataMoveType::LOGICAL; + DataMovementReason dataMoveReason = DataMovementReason::INVALID; + UID id; + try { + decodeServerKeysValue(v, assigned, emptyRange, dataMoveType, id, dataMoveReason); + } catch (Error& e) { + // Malformed serverKeys value from a partial upgrade + // or corrupted state — skip this span, don't abort + // DD init. The audit tool will report the residual + // entry via its own scan. + TraceEvent(SevWarnAlways, "DDShardEncodeRollbackSkipUndecodable", distributorId) + .detail("SS", ssId) + .detail("Key", ranges[i].key) + .detail("Error", e.code()); + continue; + } + Value oldValue = !assigned ? serverKeysFalse + : emptyRange ? serverKeysTrueEmptyRange + : serverKeysTrue; + KeyRange span = KeyRangeRef(ranges[i].key, ranges[i + 1].key); + + // One transaction per span. Retry on transient errors + // via the standard idiom. + loop { + Transaction spanTr(cx); + spanTr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); + spanTr.setOption(FDBTransactionOptions::LOCK_AWARE); + Error err; + try { + co_await krmSetRangeCoalescing(&spanTr, mapPrefix, span, allKeys, oldValue); + co_await spanTr.commit(); + break; + } catch (Error& e) { + err = e; + } + co_await spanTr.onError(err); + } + rewritesThisScan++; + } + + if (!morePages) { + break; + } + if (ranges.size() == 0) { + // morePages with size==0 shouldn't happen; guard against infinite loop. + break; + } + Key nextBegin = ranges.back().key; + // Monotonic-progress guard: if the KRM returned a page + // that doesn't advance beginKey (degenerate KRM under + // BUGGIFY KRM_GET_RANGE_LIMIT=10), stop and mark this SS + // not fully drained (fullyDrained=false below) so the + // caller leaves the sentinel unset and a later DD init + // retries rather than falsely reporting ROLLBACK COMPLETE. + if (!(beginKey < nextBegin)) { + TraceEvent(SevWarnAlways, "DDShardEncodeRollbackPhase3NoProgress", distributorId) + .detail("SS", ssId) + .detail("BeginKey", beginKey) + .detail("NextBegin", nextBegin) + .detail("PageEntries", ranges.size()); + noProgress = true; + break; + } + beginKey = nextBegin; + pageIdx++; + } + + rewritesForThisSS += rewritesThisScan; + if (rewritesThisScan == 0) { + break; // a complete scan found nothing new-format → SS drained + } + if (noProgress) { + fullyDrained = false; // bailed with possible residue; caller must not seal + break; + } + if (scanIdx >= 8) { + fullyDrained = false; // bailed with possible residue; caller must not seal + TraceEvent(SevWarnAlways, "DDShardEncodeRollbackPhase3SSScanLimit", distributorId) + .detail("SS", ssId) + .detail("Rewrites", rewritesForThisSS); + break; + } + } + + TraceEvent(SevInfo, "DDShardEncodeRollbackPhase3SSDone", distributorId) + .detail("SS", ssId) + .detail("Rewrites", rewritesForThisSS) + .detail("Scans", scanIdx) + .detail("ElapsedSec", now() - ssStart); + co_return rewritesForThisSS; + } + + // Set the shard-encode migration sentinel to `value` in its own + // transaction, retrying on transient errors. Load-bearing commit — + // audit tools and future DD inits read this key as the authoritative + // migration-complete signal. + static Future commitShardEncodeMigrationSentinel(Database cx, Value value) { + co_await runRYWTransactionVoid(cx, [value](Reference tr) -> Future { + tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); + tr->setOption(FDBTransactionOptions::LOCK_AWARE); + tr->set(shardEncodeMigrationCompleteKey, value); + return Void(); + }); + } + + static Future rewriteShardEncodedMetadata(Transaction& tr, UID distributorId, Key& phase2Cursor) { + Database cx = tr.getDatabase(); + tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::LOCK_AWARE); + + // Fast-path: sentinel says already rolled back → skip everything. + // Cost per DD init when the cluster has been at knob=false for a + // while is a single key read. + Optional marker = co_await tr.get(shardEncodeMigrationCompleteKey); + if (marker.present() && marker.get() == shardEncodeMigrationValueOld) { + TraceEvent(SevInfo, "DDShardEncodeRewriteSkipped", distributorId) + .detail("Direction", "rollback") + .detail("Reason", "SentinelSaysOld"); + co_return false; + } + TraceEvent(SevInfo, "DDShardEncodeRewriteBegin", distributorId) + .detail("Direction", "rollback") + .detail("KnobValue", SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) + .detail("PreviousSentinel", marker.present() ? marker.get() : "absent"_sr); + + // Clear the sentinel so any observer during the rewrite (audit + // tool) sees "in progress" rather than a stale "complete" + // marker. This clear is committed alongside whichever phase's + // first commit fires (or as a standalone commit before Phase 3 + // if 1 & 2 are no-ops). + tr.clear(shardEncodeMigrationCompleteKey); + + if (co_await clearShardEncodedDataMoves(tr, distributorId)) { + co_return true; // committed; caller re-enters + } + + if (co_await rewriteShardEncodedKeyServers(tr, distributorId, phase2Cursor)) { + co_return true; // committed; caller re-enters + } + + // No Phase 1 or Phase 2 work. Commit the sentinel-clear, then + // Phase 3 opens per-span transactions. Re-read serverList until + // stable to close the register-during-migration race — see + // comment before the do/while loop below. + co_await tr.commit(); + tr.reset(); + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + + // Phase 3: for each SS, rewrite its serverKeys KRM. Uses ONE + // FRESH TRANSACTION PER krmSetRangeCoalescing CALL — the same + // pattern natural DD moves use in MoveKeys.cpp:99. Correctness + // argument: krmSetRangeCoalescing uses Snapshot::True reads to + // compute coalescing boundaries; NativeAPI Transaction does not + // surface prior same-tx writes to those reads, so batching + // calls in one tx produces fragmentation (see + // KeyRangeMap.cpp:302 assertion; observed in v4/v5/v6-fix + // pre-sentinel iterations). One-tx-per-span sidesteps the RYW + // dependency entirely. + // + // Loop until the serverList is stable across a full Phase 3 + // pass. Closes the race where an SS registers between the + // snapshot and the sentinel commit — its serverKeys entries + // might otherwise be missed while sentinel="old" is set. In the + // steady case this runs once (empty second pass). + int64_t totalPhase3Rewrites = 0; + double phase3Start = now(); + std::set processed; + int passes = 0; + bool ssListStabilized = false; + // Cleared if any per-SS drain bails with possible residue (no-progress + // / scan-limit). We only seal the sentinel when every processed SS + // fully drained. + bool allDrained = true; + while (true) { + passes++; + Transaction slTr(cx); + slTr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + slTr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + slTr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + RangeResult serverList = co_await slTr.getRange(serverListKeys, CLIENT_KNOBS->TOO_MANY); + ASSERT(!serverList.more && serverList.size() < CLIENT_KNOBS->TOO_MANY); + int64_t passRewrites = 0; + int newSSCount = 0; + for (const auto& serverKv : serverList) { + UID ssId = decodeServerListValue(serverKv.value).id(); + if (!processed.insert(ssId).second) { + continue; // Already processed in a prior pass + } + newSSCount++; + bool ssDrained = true; + passRewrites += co_await rewriteOneServerKeysKRM(cx, ssId, distributorId, ssDrained); + if (!ssDrained) { + allDrained = false; + } + } + totalPhase3Rewrites += passRewrites; + if (newSSCount == 0) { + // No SSes appeared since the last pass — safe to seal. + ssListStabilized = true; + break; + } + TraceEvent(SevInfo, "DDShardEncodeRollbackPhase3Repass", distributorId) + .detail("Pass", passes) + .detail("NewSSes", newSSCount) + .detail("PassRewrites", passRewrites); + // Bounded safety net: if an operator is registering SSes + // continuously, break out after a reasonable number of + // passes and let the next DD init retry. + if (passes >= 8) { + TraceEvent(SevWarnAlways, "DDShardEncodeRollbackPhase3PassLimit", distributorId) + .detail("Passes", passes) + .detail("Reason", "SSListNotStabilizing"); + break; + } + } + + if (!ssListStabilized || !allDrained) { + // Either an SS registered after our last pass (list not stable), + // or a per-SS drain bailed with possible residue (no-progress / + // scan-limit). Leave the sentinel unset so a later DD init + // re-scans — sealing "old" here would falsely report "safe to + // downgrade". Return false (not true) so we don't immediately + // re-enter and spin on the same churn. + TraceEvent(SevWarnAlways, "DDShardEncodeRollbackPhase3Incomplete", distributorId) + .detail("Phase3Rewrites", totalPhase3Rewrites) + .detail("Phase3Passes", passes) + .detail("SSListStabilized", ssListStabilized) + .detail("AllDrained", allDrained) + .detail("Phase3ElapsedSec", now() - phase3Start); + co_return false; + } + + // All phases drained. Set sentinel = "old" so subsequent DD + // inits fast-path skip. + co_await commitShardEncodeMigrationSentinel(cx, shardEncodeMigrationValueOld); + TraceEvent(SevInfo, "DDShardEncodeRewriteComplete", distributorId) + .detail("Direction", "rollback") + .detail("Phase3Rewrites", totalPhase3Rewrites) + .detail("Phase3Passes", passes) + .detail("Phase3ElapsedSec", now() - phase3Start); + + // Return true if we did any Phase 3 work (caller restart is + // cheap — sentinel fast-path skip on re-entry). + co_return totalPhase3Rewrites > 0; + } + + // When the effective encoding target is "encoded" (new format) on DD + // init -- i.e. shard_metadata_format=encoded, or the + // SHARD_ENCODE_LOCATION_METADATA knob when the config is UNSET -- clear + // any stale "sentinel=='old'" that a prior rollback + // (rewriteShardEncodedMetadata) committed. This runs unconditionally on + // the forward path (independent of shard_metadata_migration): without + // it, natural DD moves during the forward window would write new-format + // entries while the sentinel still claims "old"; a subsequent rollback + // (target back to original) would see sentinel=="old" in + // rewriteShardEncodedMetadata's fast-path, skip the rewrite, and leave + // those new-format entries in place -- old-format DD would then trip on + // them. + // + // We do NOT actively rewrite old-format entries to new format on + // re-forward: new-format code paths (decodeServerKeysValue, + // decodeKeyServersValue) explicitly accept old-format inputs, and + // natural DD moves gradually rewrite as ranges move. Trade: audit + // tool's FORWARD COMPLETE terminal state converges over the natural- + // move timescale rather than in a bounded post-flip window. + // + // Idempotent -- safe on a fresh cluster (sentinel absent) and on + // repeat DD inits. Steady-state cost when the target is encoded: one + // system-key read per DD init, zero commits after the initial clear. + static Future clearStaleShardEncodedRewriteSentinel(Transaction& tr, UID distributorId) { + tr.setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::LOCK_AWARE); - co_return false; + Optional marker = co_await tr.get(shardEncodeMigrationCompleteKey); + if (!marker.present()) { + co_return false; + } + tr.clear(shardEncodeMigrationCompleteKey); + co_await tr.commit(); + TraceEvent(SevInfo, "DDShardEncodeSentinelCleared", distributorId).detail("PreviousValue", marker.get()); + tr.reset(); + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + tr.setOption(FDBTransactionOptions::PRIORITY_SYSTEM_IMMEDIATE); + co_return true; } // Read keyservers, return unique set of teams @@ -390,7 +775,8 @@ class DDTxnProcessorImpl { MoveKeysLock moveKeysLock, std::vector> remoteDcIds, const DDEnabledState* ddEnabledState, - SkipDDModeCheck skipDDModeCheck) { + SkipDDModeCheck skipDDModeCheck, + DatabaseConfiguration dbConfig) { auto result = makeReference(); Key beginKey = allKeys.begin; @@ -410,6 +796,27 @@ class DDTxnProcessorImpl { Optional healthyZone = co_await getHealthyZone(cx, distributorId); result->initHealthyZoneValue = healthyZone; + // Reuse DD's already-loaded DatabaseConfiguration (passed in from + // loadDatabaseConfiguration) rather than a per-init read here. + // The rewrite is gated on shard_metadata_migration below; with the + // default (UNSET) config no rewrite runs. NOTE this is an + // intentional change from the legacy behavior where the + // SHARD_ENCODE_LOCATION_METADATA knob being false alone triggered + // the rollback rewrite — the knob now only selects the target + // format as a fallback when shard_metadata_format is UNSET. + bool migrationEnabled = + dbConfig.shardMetadataMigration == DatabaseConfiguration::ShardMetadataMigration::ENABLED; + bool targetIsNewFormat; + if (dbConfig.shardMetadataFormat != DatabaseConfiguration::ShardMetadataFormat::UNSET) { + targetIsNewFormat = (dbConfig.shardMetadataFormat == DatabaseConfiguration::ShardMetadataFormat::ENCODED); + } else { + targetIsNewFormat = SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA; + } + // Phase 2 keyServers cursor — persists across re-entries so the + // rewrite paginates the whole prefix instead of rescanning from + // the head each time. + Key phase2Cursor = keyServersPrefix; + CODE_PROBE( (bool)skipDDModeCheck, "DD Mode won't prevent read initial data distribution.", probe::decoration::rare); // Get the server list in its own try/catch block since it modifies result. We don't want a subsequent failure @@ -477,10 +884,53 @@ class DDTxnProcessorImpl { } } - // If SHARD_ENCODE is off, rewrite any shard-encoded metadata to old format. - if (!SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { - if (co_await rewriteShardEncodedMetadata(tr, distributorId)) { - continue; // Committed a rewrite — re-read from the top + // Rewrite metadata that doesn't match the current effective + // format (rollback direction — actively drains new-format + // entries to old), or invalidate any stale rollback-complete + // sentinel on re-forward so a subsequent rollback doesn't + // fast-path skip incorrectly. On re-forward we rely on + // natural DD moves to eventually rewrite entries in new + // format; new-format code paths accept old-format inputs, + // so mixed state is safe. + // + // The rollback (active rewrite) branch is GATED on the two + // DatabaseConfiguration options (default UNSET), resolved once + // above the retry loop. The forward branch's stale-sentinel + // clear runs UNCONDITIONALLY (independent of + // shard_metadata_migration). Direction is implicit -- DD + // converges toward the format specified by + // shard_metadata_format. + // + // Coexistence with the legacy SHARD_ENCODE_LOCATION_METADATA + // knob: if shard_metadata_format is UNSET, DD falls back to + // the knob value (true -> encoded, false -> original) + // as the target. If shard_metadata_migration is UNSET, no + // migration runs — the knob being false no longer triggers + // a rewrite on its own (an intentional change from the + // legacy knob-only trigger). + if (!targetIsNewFormat) { + // Rollback direction: the active rewrite is opt-in via + // shard_metadata_migration. + if (migrationEnabled) { + if (co_await rewriteShardEncodedMetadata(tr, distributorId, phase2Cursor)) { + continue; // Committed a rewrite — re-read from the top + } + } + } else { + // Forward direction: ALWAYS clear a stale rollback-complete + // sentinel, independent of shard_metadata_migration. The + // forward direction has no active rewrite, but if a prior + // rollback left sentinel=="old" and the operator re-forwards + // with migration disabled, the sentinel must still be cleared + // — otherwise a later rollback reads the stale "old" via + // rewriteShardEncodedMetadata's fast-path and skips the + // rewrite, so the new-format entries written while forward + // never converge. clearStaleShardEncodedRewriteSentinel is a + // single-key read (a no-op when the sentinel is absent, i.e. + // on clusters that never rolled back), so this stays a no-op + // for never-migrated clusters. + if (co_await clearStaleShardEncodedRewriteSentinel(tr, distributorId)) { + continue; // Committed a clear — re-read from the top } } @@ -674,7 +1124,7 @@ class DDTxnProcessorImpl { .detail("NumShards", result->shards.size()) .detail("ElapsedSeconds", now() - keyServerScanStart); - if (SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA && numDataMoves > 0) { + if (targetIsNewFormat && numDataMoves > 0) { for (int shard = 0; shard < result->shards.size() - 1; ++shard) { const DDShardInfo& iShard = result->shards[shard]; KeyRangeRef keys = KeyRangeRef(iShard.key, result->shards[shard + 1].key); @@ -917,9 +1367,10 @@ Future> DDTxnProcessor::getInitialDataDistrib const MoveKeysLock& moveKeysLock, const std::vector>& remoteDcIds, const DDEnabledState* ddEnabledState, - SkipDDModeCheck skipDDModeCheck) { + SkipDDModeCheck skipDDModeCheck, + const DatabaseConfiguration& configuration) { return DDTxnProcessorImpl::getInitialDataDistribution( - cx, distributorId, moveKeysLock, remoteDcIds, ddEnabledState, skipDDModeCheck); + cx, distributorId, moveKeysLock, remoteDcIds, ddEnabledState, skipDDModeCheck, configuration); } Future DDTxnProcessor::waitForDataDistributionEnabled(const DDEnabledState* ddEnabledState) const { diff --git a/fdbserver/datadistributor/DDTxnProcessor.h b/fdbserver/datadistributor/DDTxnProcessor.h index caf216993f4..109f8c8910e 100644 --- a/fdbserver/datadistributor/DDTxnProcessor.h +++ b/fdbserver/datadistributor/DDTxnProcessor.h @@ -77,7 +77,8 @@ class IDDTxnProcessor : public ReferenceCounted { const MoveKeysLock& moveKeysLock, const std::vector>& remoteDcIds, const DDEnabledState* ddEnabledState, - SkipDDModeCheck skipDDModeCheck) = 0; + SkipDDModeCheck skipDDModeCheck, + const DatabaseConfiguration& configuration) = 0; virtual ~IDDTxnProcessor() = default; @@ -171,7 +172,8 @@ class DDTxnProcessor : public IDDTxnProcessor { const MoveKeysLock& moveKeysLock, const std::vector>& remoteDcIds, const DDEnabledState* ddEnabledState, - SkipDDModeCheck skipDDModeCheck) override; + SkipDDModeCheck skipDDModeCheck, + const DatabaseConfiguration& configuration) override; Future takeMoveKeysLock(UID const& ddId) const override; diff --git a/fdbserver/datadistributor/DataDistribution.cpp b/fdbserver/datadistributor/DataDistribution.cpp index 61cade8ea07..4cfd3b8c717 100644 --- a/fdbserver/datadistributor/DataDistribution.cpp +++ b/fdbserver/datadistributor/DataDistribution.cpp @@ -492,7 +492,8 @@ struct DataDistributor : NonCopyable, ReferenceCounted { lock, configuration.usableRegions > 1 ? remoteDcIds : std::vector>(), context->ddEnabledState.get(), - SkipDDModeCheck::False)); + SkipDDModeCheck::False, + configuration)); } void initDcInfo() { @@ -636,9 +637,27 @@ struct DataDistributor : NonCopyable, ReferenceCounted { .setMaxFieldLength(-1) .detail("Conf", self->configuration.toString()); + // Resolve the effective shard-location-metadata encoding target for + // this DD incarnation: DatabaseConfiguration is authoritative, with + // the legacy SHARD_ENCODE_LOCATION_METADATA knob as the fallback + // only when shard_metadata_format is UNSET. Publish it on + // ddEnabledState so every downstream write/move path reads one + // resolved value instead of the raw knob. + bool shardEncodeLocationMetadata = + self->configuration.shardMetadataFormatIsEncoded().orDefault( + SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + self->context->ddEnabledState->setShardEncodeLocationMetadata(shardEncodeLocationMetadata); + TraceEvent("DDInitShardEncodeTarget", self->ddId) + .detail("ShardEncodeLocationMetadata", shardEncodeLocationMetadata) + .detail("ConfigFormatUnset", + self->configuration.shardMetadataFormat == + DatabaseConfiguration::ShardMetadataFormat::UNSET) + .detail("Knob", SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + if (self->configuration.storageServerStoreType == KeyValueStoreType::SSD_SHARDED_ROCKSDB && - !SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + !shardEncodeLocationMetadata) { TraceEvent(SevError, "PhysicalShardNotEnabledForShardedRocks", self->ddId) + .detail("Reason", "sharded-rocksdb requires new-format shard-location metadata") .detail("EnableServerKnob", "SHARD_ENCODE_LOCATION_METADATA"); throw internal_error(); } @@ -769,7 +788,7 @@ struct DataDistributor : NonCopyable, ReferenceCounted { } std::vector customBoundaries; - if (bulkLoadIsEnabled(self->initData->bulkLoadMode)) { + if (bulkLoadIsEnabled(self->initData->bulkLoadMode, self->context->ddEnabledState->shardEncodeLocationMetadata())) { // Bulk load does not allow boundary change TraceEvent(SevInfo, "DDInitCustomRangeConfigDisabledByBulkLoadMode", self->ddId); } else { @@ -904,7 +923,7 @@ struct DataDistributor : NonCopyable, ReferenceCounted { .detail("DataMove", meta.toString()); cancelledMoves++; } else if (it.value()->isCancelled() || - (it.value()->valid && !SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA)) { + (it.value()->valid && !self->context->ddEnabledState->shardEncodeLocationMetadata())) { RelocateShard rs(meta.ranges.front(), DataMovementReason::RECOVER_MOVE, RelocateReason::OTHER); rs.dataMoveId = meta.id; rs.cancelled = true; @@ -2313,10 +2332,11 @@ Future monitorBulkLoadModeAndSpawnActors(Reference self, co_return; } - // Only monitor if SHARD_ENCODE_LOCATION_METADATA is enabled (required for bulkload) - if (!SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { + // Only monitor if the effective shard-location-metadata target is new + // format (required for bulkload) + if (!self->context->ddEnabledState->shardEncodeLocationMetadata()) { TraceEvent(SevInfo, "DDBulkLoadModeMonitorSkipped", self->ddId) - .detail("Reason", "SHARD_ENCODE_LOCATION_METADATA is disabled"); + .detail("Reason", "effective shard_metadata target is original"); co_return; } @@ -2341,7 +2361,8 @@ Future monitorBulkLoadModeAndSpawnActors(Reference self, rd >> mode; } - if (bulkLoadIsEnabled(mode) && !self->bulkLoadEnabled) { + if (bulkLoadIsEnabled(mode, self->context->ddEnabledState->shardEncodeLocationMetadata()) && + !self->bulkLoadEnabled) { TraceEvent(SevInfo, "DDBulkLoadModeDynamicallyEnabled", self->ddId) .detail("UsableRegions", self->configuration.usableRegions); self->bulkLoadEnabled = true; @@ -2774,7 +2795,7 @@ Future bulkDumpCore(Reference self, Future readyToS } void addDataDistributionActors(Reference self, std::vector>& actors) { - if (bulkLoadIsEnabled(self->initData->bulkLoadMode)) { + if (bulkLoadIsEnabled(self->initData->bulkLoadMode, self->context->ddEnabledState->shardEncodeLocationMetadata())) { TraceEvent(SevInfo, "DDBulkLoadModeEnabled", self->ddId) .detail("UsableRegions", self->configuration.usableRegions); self->bulkLoadEnabled = true; @@ -2892,7 +2913,9 @@ Future dataDistribution(Reference self, .anyZeroHealthyTeams = anyZeroHealthyTeams, .shards = &shards, .trackerCancelled = &self->context->trackerCancelled, - .usableRegions = self->configuration.usableRegions }); + .usableRegions = self->configuration.usableRegions, + .shardEncodeLocationMetadata = + self->context->ddEnabledState->shardEncodeLocationMetadata() }); actors.push_back(reportErrorsExcept(DataDistributionTracker::run(self->context->tracker, self->initData, getShardMetrics.getFuture(), diff --git a/fdbserver/datadistributor/DataDistribution.h b/fdbserver/datadistributor/DataDistribution.h index 1fd6bdccc8c..d488767747c 100644 --- a/fdbserver/datadistributor/DataDistribution.h +++ b/fdbserver/datadistributor/DataDistribution.h @@ -554,8 +554,8 @@ struct DDBulkLoadEngineTask { } }; -inline bool bulkLoadIsEnabled(int bulkLoadModeValue) { - return SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA && bulkLoadModeValue == 1; +inline bool bulkLoadIsEnabled(int bulkLoadModeValue, bool shardEncodeLocationMetadata) { + return shardEncodeLocationMetadata && bulkLoadModeValue == 1; } inline bool bulkDumpIsEnabled(int bulkDumpModeValue) { diff --git a/fdbserver/workloads/CheckMetadataEncoding.cpp b/fdbserver/workloads/CheckMetadataEncoding.cpp index 9069df30aeb..f9dc5637cb3 100644 --- a/fdbserver/workloads/CheckMetadataEncoding.cpp +++ b/fdbserver/workloads/CheckMetadataEncoding.cpp @@ -19,9 +19,13 @@ */ // Workload that verifies keyServers and serverKeys encoding format matches the -// SHARD_ENCODE_LOCATION_METADATA knob. Used to validate forward migration and -// rollback of the shard-encoded metadata feature. +// cluster's effective shard-location-metadata target. The effective target is +// DatabaseConfiguration's shard_metadata_format when set, otherwise the +// SHARD_ENCODE_LOCATION_METADATA knob (same resolution DD uses). Used to +// validate forward migration and rollback of the shard-encoded metadata +// feature, driven either by the knob or by `configure shard_metadata_format`. +#include "fdbclient/DatabaseConfiguration.h" #include "fdbclient/NativeAPI.actor.h" #include "fdbclient/SystemData.h" #include "fdbserver/core/Knobs.h" @@ -34,20 +38,28 @@ struct CheckMetadataEncodingWorkload : TestWorkload { bool shardEncodeExpected; bool allowMixedFormats; // True in rollback scenarios where old entries remain - bool requireKnobFalse; // If true, assert that SHARD_ENCODE is actually false - // If true (in a knob=true phase), require the FORWARD COMPLETE condition: + // If true, require the cluster's effective target to be original (i.e. + // rolled back). Works whether the rollback was driven by the knob + // (knob=false) or by configure (shard_metadata_format=original with the + // knob left as-is). Named for historical reasons; it now checks the + // resolved effective target, not the raw knob. + bool requireKnobFalse; + // If true (effective target encoded), require the FORWARD COMPLETE condition: // zero old-format assigned entries in both keyServers and serverKeys. // Uses the same counting logic as fdbcli's `audit_storage // metadata_encoding` command, so this option gates on the same // terminal state that command reports. bool requireForwardComplete; - // If true (in a knob=false phase), require the ROLLBACK COMPLETE + // If true (effective target original), require the ROLLBACK COMPLETE // condition: zero new-format entries in both keyServers and // serverKeys, AND no dataMoves entries. Same condition the fdbcli // audit tool uses to report "safe to downgrade binary". bool requireRollbackComplete; explicit CheckMetadataEncodingWorkload(WorkloadContext const& wcx) : TestWorkload(wcx) { + // Default before the effective target is resolved from config in _start; + // overwritten there. The knob is only the fallback when + // shard_metadata_format is UNSET. shardEncodeExpected = SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA; allowMixedFormats = getOption(options, "allowMixedFormats"_sr, false); requireKnobFalse = getOption(options, "requireKnobFalse"_sr, false); @@ -55,17 +67,7 @@ struct CheckMetadataEncodingWorkload : TestWorkload { requireRollbackComplete = getOption(options, "requireRollbackComplete"_sr, false); } - Future setup(Database const& cx) override { - if (requireKnobFalse && SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) { - TraceEvent(SevError, "CheckMetadataEncodingKnobNotFalse") - .detail("Expected", false) - .detail("Actual", SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) - .detail("Hint", - "TOML [[knobs]] override for shard_encode_location_metadata " - "did not take effect. The knob infrastructure fix may not be working."); - } - return Void(); - } + Future setup(Database const& cx) override { return Void(); } Future start(Database const& cx) override { if (clientId != 0) return Void(); @@ -75,111 +77,216 @@ struct CheckMetadataEncodingWorkload : TestWorkload { Future check(Database const& cx) override { return true; } void getMetrics(std::vector& m) override {} + // Resolve the cluster's effective shard-location-metadata target the same + // way DD does: DatabaseConfiguration.shard_metadata_format when set, + // otherwise the SHARD_ENCODE_LOCATION_METADATA knob. + static Future resolveEffectiveTarget(Database cx) { + while (true) { + Transaction tr(cx); + Error err; + try { + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + Optional v = co_await tr.get( + StringRef(DatabaseConfiguration::SHARD_METADATA_FORMAT_KEY).withPrefix(configKeysPrefix)); + if (v.present()) { + co_return v.get() == StringRef(DatabaseConfiguration::SHARD_METADATA_FORMAT_ENCODED); + } + co_return SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA; + } catch (Error& e) { + err = e; + } + co_await tr.onError(err); + } + } + Future _start(CheckMetadataEncodingWorkload* self, Database cx) { + // Resolve the effective target (config-or-knob) and drive all + // expectations off it, so the workload validates both knob-driven and + // configure-driven rollback/forward. + self->shardEncodeExpected = co_await resolveEffectiveTarget(cx); + if (self->requireKnobFalse && self->shardEncodeExpected) { + TraceEvent(SevError, "CheckMetadataEncodingEffectiveNotOldFormat") + .detail("Expected", "original") + .detail("EffectiveIsNewFormat", self->shardEncodeExpected) + .detail("Knob", SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA) + .detail("Hint", + "requireKnobFalse set but the effective shard_metadata target is still " + "encoded. The knob override or `configure shard_metadata_format=original` " + "did not take effect."); + } + + // When a terminal-state condition is required, the DD-side rewrite + // (forward migration or rollback) runs asynchronously and may not be + // finished the instant this workload starts scanning. Poll until the + // required terminal condition holds or a deadline elapses, then + // assert on the final observation. This removes the false failure + // where the audit scanned a few seconds before the DD rewrite sealed + // (observed at T=249 while the rewrite completed at T=253 — see + // journal 2026-07-18). Only poll when the requested terminal mode + // matches the knob direction; a mismatched request is a test + // misconfiguration and should fail loudly and immediately below. + const bool pollForConvergence = (self->requireRollbackComplete && !self->shardEncodeExpected) || + (self->requireForwardComplete && self->shardEncodeExpected); + // Deadline is generous: under BUGGIFY the DD rollback rewrite can be + // throttled hard (tiny SHARD_ENCODE_REWRITE_KS_BATCH_SIZE and + // KRM_GET_RANGE_LIMIT), making Phase 2/3 legitimately take a few + // minutes of sim time to drain a large shard set. Too short a + // deadline produces a false timeout while DD is still correctly + // converging (observed at ~182s with both knobs pinned small). + const double pollDeadline = now() + 400.0; + const double pollInterval = 2.0; + int64_t keyServersOld = 0, keyServersNew = 0; int64_t serverKeysOld = 0, serverKeysNew = 0; int64_t dataMovesCount = 0; + bool dataMovesScanned = false; + int scanAttempts = 0; - // Scan keyServers. - // Empty-value entries are KRM boundary sentinels — format-neutral, - // skipped for the same reason the fdbcli audit tool skips them - // (see fdbcli/CheckMetadataEncodingCommand.cpp). - { - Key begin = keyServersPrefix; - Key end = keyServersEnd; - while (begin < end) { - Transaction tr(cx); - tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); - tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); - Error err; - try { - RangeResult result = co_await tr.getRange(KeyRangeRef(begin, end), 1000); - for (const auto& kv : result) { - if (kv.value.empty()) { - continue; + while (true) { + scanAttempts++; + keyServersOld = keyServersNew = 0; + serverKeysOld = serverKeysNew = 0; + dataMovesCount = 0; + dataMovesScanned = false; + + // Scan keyServers. + // Empty-value entries are KRM boundary sentinels — format-neutral, + // skipped for the same reason the fdbcli audit tool skips them + // (see fdbcli/CheckMetadataEncodingCommand.cpp). + { + Key begin = keyServersPrefix; + Key end = keyServersEnd; + while (begin < end) { + Transaction tr(cx); + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + Error err; + try { + RangeResult result = co_await tr.getRange(KeyRangeRef(begin, end), 1000); + for (const auto& kv : result) { + if (kv.value.empty()) { + continue; + } + BinaryReader rd(kv.value, IncludeVersion()); + if (rd.protocolVersion().hasShardEncodeLocationMetaData()) { + keyServersNew++; + } else { + keyServersOld++; + } } - BinaryReader rd(kv.value, IncludeVersion()); - if (rd.protocolVersion().hasShardEncodeLocationMetaData()) { - keyServersNew++; - } else { - keyServersOld++; + if (!result.more) { + break; } + begin = keyAfter(result.back().key); + } catch (Error& e) { + err = e; } - if (!result.more) { - break; + if (err.isValid()) { + co_await tr.onError(err); } - begin = keyAfter(result.back().key); - } catch (Error& e) { - err = e; - } - if (err.isValid()) { - co_await tr.onError(err); } } - } - // Scan serverKeys. - // Uses the same classifiers as the fdbcli audit tool: format- - // neutral entries (empty sentinels + serverKeysFalse) are - // skipped, old-format assignments (serverKeysTrue / - // serverKeysTrueEmptyRange) and new-format assignments - // (UID-encoded) are counted separately. - { - Key begin = serverKeysPrefix; - Key end = strinc(serverKeysPrefix); - while (begin < end) { - Transaction tr(cx); - tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); - tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); - Error err; - try { - RangeResult result = co_await tr.getRange(KeyRangeRef(begin, end), 1000); - for (const auto& kv : result) { - if (isServerKeysUnassigned(kv.value)) { - continue; + // Scan serverKeys. + // Uses the same classifiers as the fdbcli audit tool: format- + // neutral entries (empty sentinels + serverKeysFalse) are + // skipped, old-format assignments (serverKeysTrue / + // serverKeysTrueEmptyRange) and new-format assignments + // (UID-encoded) are counted separately. + { + Key begin = serverKeysPrefix; + Key end = strinc(serverKeysPrefix); + while (begin < end) { + Transaction tr(cx); + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + Error err; + try { + RangeResult result = co_await tr.getRange(KeyRangeRef(begin, end), 1000); + for (const auto& kv : result) { + if (isServerKeysUnassigned(kv.value)) { + continue; + } + if (isServerKeysOldFormatAssigned(kv.value)) { + serverKeysOld++; + } else { + serverKeysNew++; + } } - if (isServerKeysOldFormatAssigned(kv.value)) { - serverKeysOld++; - } else { - serverKeysNew++; + if (!result.more) { + break; } + begin = keyAfter(result.back().key); + } catch (Error& e) { + err = e; } - if (!result.more) { - break; + if (err.isValid()) { + co_await tr.onError(err); } - begin = keyAfter(result.back().key); - } catch (Error& e) { - err = e; } - if (err.isValid()) { + } + + // Count in-flight data moves. Needed for the ROLLBACK COMPLETE + // terminal condition (same as the fdbcli audit tool). Only scan + // when the assertion actually needs the count — the scan is + // otherwise wasted I/O and the trace event below distinguishes + // "not scanned" from "0 entries" by omitting the DataMoves detail. + if (self->requireRollbackComplete) { + while (true) { + Transaction tr(cx); + tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); + tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); + Error err; + try { + RangeResult result = co_await tr.getRange(dataMoveKeys, CLIENT_KNOBS->TOO_MANY); + ASSERT(!result.more && result.size() < CLIENT_KNOBS->TOO_MANY); + dataMovesCount = result.size(); + dataMovesScanned = true; + break; + } catch (Error& e) { + err = e; + } co_await tr.onError(err); } } - } - // Count in-flight data moves. Needed for the ROLLBACK COMPLETE - // terminal condition (same as the fdbcli audit tool). Only scan - // when the assertion actually needs the count — the scan is - // otherwise wasted I/O and the trace event below distinguishes - // "not scanned" from "0 entries" by omitting the DataMoves detail. - bool dataMovesScanned = false; - if (self->requireRollbackComplete) { - while (true) { - Transaction tr(cx); - tr.setOption(FDBTransactionOptions::READ_SYSTEM_KEYS); - tr.setOption(FDBTransactionOptions::READ_LOCK_AWARE); - Error err; - try { - RangeResult result = co_await tr.getRange(dataMoveKeys, CLIENT_KNOBS->TOO_MANY); - ASSERT(!result.more && result.size() < CLIENT_KNOBS->TOO_MANY); - dataMovesCount = result.size(); - dataMovesScanned = true; - break; - } catch (Error& e) { - err = e; - } - co_await tr.onError(err); + if (!pollForConvergence) { + break; // single-shot: preserve legacy one-scan behavior + } + // Terminal condition matching the requested mode. Both may be + // requested in principle; require whichever are set. + bool terminalReached = true; + if (self->requireRollbackComplete) { + terminalReached = + terminalReached && (keyServersNew == 0 && serverKeysNew == 0 && dataMovesCount == 0); + } + if (self->requireForwardComplete) { + terminalReached = terminalReached && (keyServersOld == 0 && serverKeysOld == 0); + } + if (terminalReached) { + TraceEvent(SevInfo, "CheckMetadataEncodingConverged") + .detail("ScanAttempts", scanAttempts) + .detail("KeyServersOld", keyServersOld) + .detail("KeyServersNew", keyServersNew) + .detail("ServerKeysOld", serverKeysOld) + .detail("ServerKeysNew", serverKeysNew) + .detail("DataMoves", dataMovesCount); + break; + } + if (now() >= pollDeadline) { + // Fall through to the assertions below, which fail loudly + // with the final counts. + TraceEvent(SevWarnAlways, "CheckMetadataEncodingConvergeTimeout") + .detail("ScanAttempts", scanAttempts) + .detail("KeyServersOld", keyServersOld) + .detail("KeyServersNew", keyServersNew) + .detail("ServerKeysOld", serverKeysOld) + .detail("ServerKeysNew", serverKeysNew) + .detail("DataMoves", dataMovesCount); + break; } + co_await delay(pollInterval); } auto resultEvent = TraceEvent("CheckMetadataEncodingResult"); @@ -188,6 +295,7 @@ struct CheckMetadataEncodingWorkload : TestWorkload { .detail("KeyServersNew", keyServersNew) .detail("ServerKeysOld", serverKeysOld) .detail("ServerKeysNew", serverKeysNew) + .detail("ScanAttempts", scanAttempts) .detail("RequireForwardComplete", self->requireForwardComplete) .detail("RequireRollbackComplete", self->requireRollbackComplete); if (dataMovesScanned) { @@ -236,7 +344,7 @@ struct CheckMetadataEncodingWorkload : TestWorkload { if (self->requireForwardComplete) { if (!self->shardEncodeExpected) { TraceEvent(SevError, "CheckMetadataEncodingFailed") - .detail("Reason", "requireForwardComplete set but SHARD_ENCODE_LOCATION_METADATA is false") + .detail("Reason", "requireForwardComplete set but effective shard_metadata target is original") .detail("ShardEncodeExpected", self->shardEncodeExpected); } else if (keyServersOld != 0 || serverKeysOld != 0) { // FORWARD COMPLETE: no old-format assigned entries remain. @@ -253,7 +361,7 @@ struct CheckMetadataEncodingWorkload : TestWorkload { if (self->requireRollbackComplete) { if (self->shardEncodeExpected) { TraceEvent(SevError, "CheckMetadataEncodingFailed") - .detail("Reason", "requireRollbackComplete set but SHARD_ENCODE_LOCATION_METADATA is true") + .detail("Reason", "requireRollbackComplete set but effective shard_metadata target is encoded") .detail("ShardEncodeExpected", self->shardEncodeExpected); } else if (keyServersNew != 0 || serverKeysNew != 0 || dataMovesCount != 0) { // ROLLBACK COMPLETE: no new-format entries and no data diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 72d5f3f5cd7..01d61236f0e 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -190,6 +190,7 @@ if(WITH_PYTHON) add_fdb_test(TEST_FILES fast/CheckMetadataEncodingForward.toml) add_fdb_test(TEST_FILES fast/CheckMetadataEncodingOldPath.toml) add_fdb_test(TEST_FILES fast/ShardEncodeRollback.toml) + add_fdb_test(TEST_FILES fast/ShardEncodeRollbackConfig.toml) add_fdb_test(TEST_FILES fast/MutationLogReaderCorrectness.toml) add_fdb_test(TEST_FILES fast/GetEstimatedRangeSize.toml) diff --git a/tests/fast/ShardEncodeRollback.toml b/tests/fast/ShardEncodeRollback.toml index 66b7a85e061..7878412d4d5 100644 --- a/tests/fast/ShardEncodeRollback.toml +++ b/tests/fast/ShardEncodeRollback.toml @@ -1,12 +1,41 @@ # Exclude sharded RocksDB (type 5) — it requires SHARD_ENCODE=true and # cannot operate after the knob flips to false. +# +# NOTE: shard_metadata_migration is NOT set in the initial config below. +# Enabling migration at cluster-init time caused joshua to time out during +# `configure new` under fault injection (Unable to set starting configuration). +# Instead, a ChangeConfig workload enables migration at the start of the +# ShardEncodeRollbackVerify phase (below), right when the knob flip to +# false takes effect and we want DD to run the rewrite. +# +# `datacenters = 1` + `generateFearless = false` pin the sim to a single +# datacenter, single-region setup. Without them, setRandomConfig() picks +# randInt(1,4) DCs plus multi-region layouts on top of our config, and +# joshua's fault injection struggles to bring those up (setup times out +# at 2000s with "Unable to set starting configuration"). PR2's rewrite +# logic doesn't care about regions; the extra sim variation is pure +# flakiness for this test. [configuration] storageEngineExcludeTypes=[3,4,5] +config = "triple ssd-2" +datacenters = 1 +generateFearless = false # Top-level: SHARD_ENCODE=false is the "final" state. # Block 1 overrides to true temporarily. +# +# Force fine-grained sharding so the rollback rewrite has many keyServers +# entries to convert (exercises Phase 2 pagination and, with the buggified +# SHARD_ENCODE_REWRITE_KS_BATCH_SIZE / KRM_GET_RANGE_LIMIT, multi-page / +# multi-scan draining). Small min_shard_bytes + shard_bytes_per_sqrt_bytes=0 +# keeps DD from coalescing shards back together. These values mirror +# DDPipelineSaturation.toml ("forces many small shards"): min_shard_bytes +# =10000 with a high-TPS Cycle at nodeCount=60000. Verify the resulting +# shard count via CheckMetadataEncodingResult.KeyServersOld/New in the trace. [[knobs]] shard_encode_location_metadata = false +min_shard_bytes = 10000 +shard_bytes_per_sqrt_bytes = 0 [[test]] testTitle = 'ShardEncodeRollbackSetup' @@ -17,13 +46,20 @@ testTitle = 'ShardEncodeRollbackSetup' [[test.workload]] testName = 'Cycle' - transactionsPerSecond = 2500.0 - nodeCount = 2500 - testDuration = 10.0 + transactionsPerSecond = 5000.0 + nodeCount = 60000 + testDuration = 30.0 + # RandomMoveKeys here (knob=true) builds the new-format keyServers/ + # serverKeys state that the Verify phase rolls back. It is deliberately + # confined to this phase — see the note in ShardEncodeRollbackVerify + # for why it must not run concurrently with the rollback rewrite. This + # phase is all knob=true (no rollback, no lock-contention race), so we + # run it long enough to build a large multi-page new-format state that + # stresses Phase 2/Phase 3 pagination during the rollback below. [[test.workload]] testName = 'RandomMoveKeys' - testDuration = 10.0 + testDuration = 30.0 [[test]] testTitle = 'ShardEncodeRollbackVerify' @@ -31,16 +67,39 @@ testTitle = 'ShardEncodeRollbackVerify' # Phase 2: No per-test override — inherits top-level false. # DD detects knob change, restarts, exercises rollback path. + # Enable the DD-init active rewrite via ChangeConfig at the start of + # this phase (rather than at cluster-init time, which caused + # setup-timeout failures under joshua's fault injection). Small delay + # so this fires after DD has settled into the knob=false state. [[test.workload]] - testName = 'Cycle' - transactionsPerSecond = 2500.0 - nodeCount = 2500 - testDuration = 30.0 + testName = 'ChangeConfig' + configMode = 'shard_metadata_migration=enabled' + minDelayBeforeChange = 1.0 + maxDelayBeforeChange = 3.0 [[test.workload]] - testName = 'RandomMoveKeys' + testName = 'Cycle' + transactionsPerSecond = 5000.0 + nodeCount = 60000 testDuration = 30.0 + # NOTE: RandomMoveKeys is intentionally NOT run in this rollback phase. + # It takes the moveKeysLock itself and issues moveKeys() directly, + # contending with DD for the lock. During the knob flip, one of those + # direct moves can sample shard_encode_location_metadata=true and + # commit a new-format keyServers entry in the window between DD's + # Phase 2 sweep and its seal (DD does not hold the lock across that + # gap) — DD then seals sentinel=old and the fast-path skip never + # reconverts it, so the audit never reaches ROLLBACK COMPLETE. This is + # NOT a production scenario: in production DD is the sole moveKeysLock + # holder during its init (its queue is off), prior-gen moves are + # lock-fenced, and under knob=false every dispatch path writes + # old-format, so the seal is safe. RandomMoveKeys runs in the Setup + # phase (knob=true) above to build the new-format state we roll back; + # Cycle + Attrition + Rollback below still stress the rewrite with + # traffic, DD restarts, and TLog recovery (all correctly old-format + # under knob=false). + [[test.workload]] testName = 'Attrition' machinesToKill = 1 @@ -58,3 +117,19 @@ testTitle = 'ShardEncodeRollbackVerify' testName = 'CheckMetadataEncoding' requireKnobFalse = true allowMixedFormats = true + +# Phase 3: after the churn workloads have completed, assert the audit +# tool's ROLLBACK COMPLETE terminal condition — zero new-format entries +# in keyServers/serverKeys and no in-flight dataMoves. Without PR2's +# proactive serverKeys rewrite (Phase 3 of rewriteShardEncodedMetadata), +# this would fail: new-format serverKeys entries would persist +# indefinitely on quiescent SSes and ROLLBACK COMPLETE would never be +# reached. This phase runs in its own test block so it starts after all +# concurrent churn workloads in ShardEncodeRollbackVerify have exited. +[[test]] +testTitle = 'ShardEncodeRollbackFinal' + + [[test.workload]] + testName = 'CheckMetadataEncoding' + requireKnobFalse = true + requireRollbackComplete = true diff --git a/tests/fast/ShardEncodeRollbackConfig.toml b/tests/fast/ShardEncodeRollbackConfig.toml new file mode 100644 index 00000000000..2f77af83a9e --- /dev/null +++ b/tests/fast/ShardEncodeRollbackConfig.toml @@ -0,0 +1,118 @@ +# Config-driven rollback variant of ShardEncodeRollback.toml. +# +# The sibling ShardEncodeRollback.toml drives the rollback by flipping the +# SHARD_ENCODE_LOCATION_METADATA *knob* true->false. This test instead leaves +# the knob TRUE for the entire run and drives the rollback purely through +# DatabaseConfiguration: `configure shard_metadata_format=original`. It +# proves the config-only path — DD resolves its effective encoding target from +# shard_metadata_format (knob only as fallback), so a config change alone +# (no process restart / no knob flip) rolls the cluster back to old format and +# reaches ROLLBACK COMPLETE. +# +# Exclude sharded RocksDB (type 5) — it requires new-format metadata and +# cannot operate once the effective target is original. +# +# datacenters=1 + generateFearless=false pin the sim to single-DC single-region +# (see ShardEncodeRollback.toml for the rationale). +[configuration] +storageEngineExcludeTypes=[3,4,5] +config = "triple ssd-2" +datacenters = 1 +generateFearless = false + +# The knob stays TRUE for the whole test (never flipped). Rollback is driven by +# config below. Fine-grained sharding so the rollback rewrite has many +# keyServers/serverKeys entries to convert. +[[knobs]] +shard_encode_location_metadata = true +min_shard_bytes = 10000 +shard_bytes_per_sqrt_bytes = 0 + +[[test]] +testTitle = 'ShardEncodeConfigRollbackSetup' + + # Phase 1: knob=true (inherited), config unset -> effective = encoded. + # Build shard-encoded (new-format) keyServers/serverKeys state via data + # moves that the config-driven rollback below will convert. + [[test.workload]] + testName = 'Cycle' + transactionsPerSecond = 5000.0 + nodeCount = 60000 + testDuration = 30.0 + + # RandomMoveKeys here (effective encoded) builds the new-format state the + # rollback converts. Confined to this phase — it takes the moveKeysLock + # itself and must not run concurrently with the rollback rewrite (see the + # note in ShardEncodeRollback.toml). + [[test.workload]] + testName = 'RandomMoveKeys' + testDuration = 30.0 + +# Phase 2: trigger the rollback via CONFIG ONLY — its OWN test block, so it +# fully commits before the verify/check block below. This ordering is load- +# bearing: CheckMetadataEncoding resolves the effective target (config, else +# knob) when it starts, and the knob is still true here — so the config MUST be +# set before any check runs, otherwise the check resolves effective=encoded +# and fails. (Workloads within a single [[test]] block run concurrently; test +# blocks run sequentially. An earlier version had ChangeConfig in the same +# block as the check and lost the race under joshua fault injection.) +# +# Setting shard_metadata_format=original makes DD's effective target +# original (the knob is still true but is now only the fallback, which is not +# consulted because the config is set). shard_metadata_migration=enabled turns +# on the DD-init active rewrite. The configure triggers a recovery, so the +# newly-elected DD re-resolves the target from config and runs the rewrite. +[[test]] +testTitle = 'ShardEncodeConfigRollbackTrigger' + + [[test.workload]] + testName = 'ChangeConfig' + configMode = 'shard_metadata_format=original shard_metadata_migration=enabled' + minDelayBeforeChange = 1.0 + maxDelayBeforeChange = 3.0 + +[[test]] +testTitle = 'ShardEncodeConfigRollbackVerify' + + # Phase 3: config is now original (set in the prior block). Stress the + # rewrite with traffic, DD restarts, and TLog recovery. + [[test.workload]] + testName = 'Cycle' + transactionsPerSecond = 5000.0 + nodeCount = 60000 + testDuration = 30.0 + + # RandomMoveKeys intentionally NOT run here (same reason as + # ShardEncodeRollback.toml — a direct move can race DD's rewrite seal and + # reintroduce a new-format entry). + [[test.workload]] + testName = 'Attrition' + machinesToKill = 1 + machinesToLeave = 3 + reboot = true + testDuration = 30.0 + + # Force TLog recovery to verify mixed-format metadata survives recovery. + [[test.workload]] + testName = 'Rollback' + meanDelay = 15.0 + testDuration = 30.0 + + # requireKnobFalse here means "require the effective target to be + # original" (the workload resolves effective from config, not the raw + # knob) — so it passes even though the knob is still true. + [[test.workload]] + testName = 'CheckMetadataEncoding' + requireKnobFalse = true + allowMixedFormats = true + +# Phase 4: assert ROLLBACK COMPLETE — zero new-format entries in +# keyServers/serverKeys and no in-flight dataMoves — proving the config-driven +# rollback fully drained. +[[test]] +testTitle = 'ShardEncodeConfigRollbackFinal' + + [[test.workload]] + testName = 'CheckMetadataEncoding' + requireKnobFalse = true + requireRollbackComplete = true From 3c1a0ef03830f6773394f0e6c61722748ed38bc9 Mon Sep 17 00:00:00 2001 From: michael stack Date: Fri, 24 Jul 2026 17:29:09 -0700 Subject: [PATCH 2/2] Formatting --- design/shard-encode-location-metadata.md | 8 ++-- fdbcli/CheckMetadataEncodingCommand.cpp | 7 ++-- .../include/fdbclient/DatabaseConfiguration.h | 4 +- .../clustercontroller/ClusterRecovery.cpp | 10 ++--- fdbserver/core/MoveKeys.cpp | 31 +++++++++------ fdbserver/datadistributor/DDTxnProcessor.h | 13 ++++--- .../datadistributor/DataDistribution.cpp | 38 +++++++++---------- fdbserver/workloads/CheckMetadataEncoding.cpp | 5 +-- 8 files changed, 60 insertions(+), 56 deletions(-) diff --git a/design/shard-encode-location-metadata.md b/design/shard-encode-location-metadata.md index eb31069e346..021b185778d 100644 --- a/design/shard-encode-location-metadata.md +++ b/design/shard-encode-location-metadata.md @@ -343,7 +343,7 @@ zero-additional-work operation from the cluster's perspective: The only thing this mixed state prevents is **downgrading the FDB binary**. Old FDB versions cannot decode new-format `serverKeys` -values and will misbehave. To make downgrade safe, all entries must +values and will misbehave/crash. To make downgrade safe, all entries must be drained to old format — reflected in `audit_storage metadata_encoding` returning `ROLLBACK COMPLETE — safe to downgrade binary`. @@ -461,9 +461,9 @@ When migration is enabled, the safety properties below apply: The rollback procedure (config-driven, no knob flip, no restart): 1. `fdbcli> configure shard_metadata_format=original shard_metadata_migration=enabled` -2. Force DD to re-init so it picks up the new configuration - immediately (the configure in step 1 already triggers a - recovery; this only expedites it): +2. Check step 1 caused DD to reinit, just in case. If not, force DD to re-init so it + picks up the new configuration immediately (the configure in step 1 triggers a + recovery; this should not be needed): fdbcli> datadistribution off fdbcli> datadistribution on diff --git a/fdbcli/CheckMetadataEncodingCommand.cpp b/fdbcli/CheckMetadataEncodingCommand.cpp index e7706b3786e..c86668667a9 100644 --- a/fdbcli/CheckMetadataEncodingCommand.cpp +++ b/fdbcli/CheckMetadataEncodingCommand.cpp @@ -149,9 +149,10 @@ Future checkMetadataEncodingCommandActor(Database cx, std::vector 0 || serverKeysNew > 0) { - fmt::println("Migration status: MIGRATION IN PROGRESS (mixed format: {} encoded keyServers, {} encoded serverKeys)", - keyServersNew, - serverKeysNew); + fmt::println( + "Migration status: MIGRATION IN PROGRESS (mixed format: {} encoded keyServers, {} encoded serverKeys)", + keyServersNew, + serverKeysNew); } else { fmt::println("Migration status: NOT STARTED (all original format)"); } diff --git a/fdbclient/include/fdbclient/DatabaseConfiguration.h b/fdbclient/include/fdbclient/DatabaseConfiguration.h index ee270d709fc..dd4807847f1 100644 --- a/fdbclient/include/fdbclient/DatabaseConfiguration.h +++ b/fdbclient/include/fdbclient/DatabaseConfiguration.h @@ -261,9 +261,7 @@ struct DatabaseConfiguration { return shardMetadataFormat == ShardMetadataFormat::ENCODED; } // Whether DD should actively converge existing entries at init. - bool shardMetadataMigrationEnabled() const { - return shardMetadataMigration == ShardMetadataMigration::ENABLED; - } + bool shardMetadataMigrationEnabled() const { return shardMetadataMigration == ShardMetadataMigration::ENABLED; } // Storage Migration Type StorageMigrationType storageMigrationType; diff --git a/fdbserver/clustercontroller/ClusterRecovery.cpp b/fdbserver/clustercontroller/ClusterRecovery.cpp index e0bdae2335e..701b42774ff 100644 --- a/fdbserver/clustercontroller/ClusterRecovery.cpp +++ b/fdbserver/clustercontroller/ClusterRecovery.cpp @@ -1898,11 +1898,11 @@ Future clusterRecoveryCore(Reference self) { } else { // Recruit and seed initial shard servers // This transaction must be the very first one in the database (version 1) - seedShardServers(recoveryCommitRequest.arena, - tr, - seedServers, - self->configuration.shardMetadataFormatIsEncoded().orDefault( - SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA)); + seedShardServers( + recoveryCommitRequest.arena, + tr, + seedServers, + self->configuration.shardMetadataFormatIsEncoded().orDefault(SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA)); } // initialConfChanges have not been conflict checked against any earlier writes in the recovery transaction, so do // this as early as possible in the recovery transaction but see above comments as to why it can't be absolutely diff --git a/fdbserver/core/MoveKeys.cpp b/fdbserver/core/MoveKeys.cpp index c781aae90bc..26c41e3be13 100644 --- a/fdbserver/core/MoveKeys.cpp +++ b/fdbserver/core/MoveKeys.cpp @@ -2180,8 +2180,13 @@ static Future startMoveShards(Database occ, if (SERVER_KNOBS->AUDIT_DATAMOVE_PRE_CHECK && runPreCheck) { std::vector servers(src.size() + dest.size()); std::merge(src.begin(), src.end(), dest.begin(), dest.end(), servers.begin()); - co_await auditLocationMetadataPreCheck( - occ, &tr, rangeIntersectKeys, servers, "startMoveShards_precheck", dataMoveId, ddEnabledState); + co_await auditLocationMetadataPreCheck(occ, + &tr, + rangeIntersectKeys, + servers, + "startMoveShards_precheck", + dataMoveId, + ddEnabledState); } if (destId.isValid()) { @@ -2472,9 +2477,9 @@ static Future decodeAndPreCheckShards(Database occ, bool runPreCheck, DataMoveMetaData const& dataMove, UID relocationIntervalId, - Severity sevDm, - bool* cancelDataMove, - const DDEnabledState* ddEnabledState) { + Severity sevDm, + bool* cancelDataMove, + const DDEnabledState* ddEnabledState) { std::vector completeSrc; std::unordered_set allServers; @@ -3830,8 +3835,13 @@ Future cleanUpDataMoveCore(Database occ, if (SERVER_KNOBS->AUDIT_DATAMOVE_PRE_CHECK && runPreCheck) { std::vector servers(src.size() + dest.size()); std::merge(src.begin(), src.end(), dest.begin(), dest.end(), servers.begin()); - co_await auditLocationMetadataPreCheck( - occ, &tr, rangeIntersectKeys, servers, "cleanUpDataMoveCore_precheck", dataMoveId, ddEnabledState); + co_await auditLocationMetadataPreCheck(occ, + &tr, + rangeIntersectKeys, + servers, + "cleanUpDataMoveCore_precheck", + dataMoveId, + ddEnabledState); } for (const auto& uid : src) { @@ -3873,11 +3883,8 @@ Future cleanUpDataMoveCore(Database occ, Value cleanupKsValue = ddEnabledState->shardEncodeLocationMetadata() ? keyServersValue(src, {}, srcId, UID()) : keyServersValue(UIDtoTagMap, src, {}); - krmSetPreviouslyEmptyRange(&tr, - keyServersPrefix, - rangeIntersectKeys, - cleanupKsValue, - currentShards[i + 1].value); + krmSetPreviouslyEmptyRange( + &tr, keyServersPrefix, rangeIntersectKeys, cleanupKsValue, currentShards[i + 1].value); } if (range.end == dataMove.ranges.front().end) { diff --git a/fdbserver/datadistributor/DDTxnProcessor.h b/fdbserver/datadistributor/DDTxnProcessor.h index 109f8c8910e..78666bf8cc9 100644 --- a/fdbserver/datadistributor/DDTxnProcessor.h +++ b/fdbserver/datadistributor/DDTxnProcessor.h @@ -168,12 +168,13 @@ class DDTxnProcessor : public IDDTxnProcessor { // Call NativeAPI implementation directly Future getServerListAndProcessClasses() override; - Future> getInitialDataDistribution(const UID& distributorId, - const MoveKeysLock& moveKeysLock, - const std::vector>& remoteDcIds, - const DDEnabledState* ddEnabledState, - SkipDDModeCheck skipDDModeCheck, - const DatabaseConfiguration& configuration) override; + Future> getInitialDataDistribution( + const UID& distributorId, + const MoveKeysLock& moveKeysLock, + const std::vector>& remoteDcIds, + const DDEnabledState* ddEnabledState, + SkipDDModeCheck skipDDModeCheck, + const DatabaseConfiguration& configuration) override; Future takeMoveKeysLock(UID const& ddId) const override; diff --git a/fdbserver/datadistributor/DataDistribution.cpp b/fdbserver/datadistributor/DataDistribution.cpp index 4cfd3b8c717..4f05f78a1a2 100644 --- a/fdbserver/datadistributor/DataDistribution.cpp +++ b/fdbserver/datadistributor/DataDistribution.cpp @@ -643,15 +643,13 @@ struct DataDistributor : NonCopyable, ReferenceCounted { // only when shard_metadata_format is UNSET. Publish it on // ddEnabledState so every downstream write/move path reads one // resolved value instead of the raw knob. - bool shardEncodeLocationMetadata = - self->configuration.shardMetadataFormatIsEncoded().orDefault( - SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); + bool shardEncodeLocationMetadata = self->configuration.shardMetadataFormatIsEncoded().orDefault( + SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); self->context->ddEnabledState->setShardEncodeLocationMetadata(shardEncodeLocationMetadata); TraceEvent("DDInitShardEncodeTarget", self->ddId) .detail("ShardEncodeLocationMetadata", shardEncodeLocationMetadata) .detail("ConfigFormatUnset", - self->configuration.shardMetadataFormat == - DatabaseConfiguration::ShardMetadataFormat::UNSET) + self->configuration.shardMetadataFormat == DatabaseConfiguration::ShardMetadataFormat::UNSET) .detail("Knob", SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA); if (self->configuration.storageServerStoreType == KeyValueStoreType::SSD_SHARDED_ROCKSDB && @@ -788,7 +786,8 @@ struct DataDistributor : NonCopyable, ReferenceCounted { } std::vector customBoundaries; - if (bulkLoadIsEnabled(self->initData->bulkLoadMode, self->context->ddEnabledState->shardEncodeLocationMetadata())) { + if (bulkLoadIsEnabled(self->initData->bulkLoadMode, + self->context->ddEnabledState->shardEncodeLocationMetadata())) { // Bulk load does not allow boundary change TraceEvent(SevInfo, "DDInitCustomRangeConfigDisabledByBulkLoadMode", self->ddId); } else { @@ -2902,20 +2901,19 @@ Future dataDistribution(Reference self, actors.push_back(self->pollMoveKeysLock()); actors.push_back(monitorBackupPartitionRequired(self->txnProcessor->context(), &shards, self->ddId)); - self->context->tracker = makeReference( - DataDistributionTrackerInitParams{ .db = self->txnProcessor, - .distributorId = self->ddId, - .readyToStart = self->initialized, - .output = self->relocationProducer, - .shardsAffectedByTeamFailure = self->shardsAffectedByTeamFailure, - .physicalShardCollection = self->physicalShardCollection, - .bulkLoadTaskCollection = self->bulkLoadTaskCollection, - .anyZeroHealthyTeams = anyZeroHealthyTeams, - .shards = &shards, - .trackerCancelled = &self->context->trackerCancelled, - .usableRegions = self->configuration.usableRegions, - .shardEncodeLocationMetadata = - self->context->ddEnabledState->shardEncodeLocationMetadata() }); + self->context->tracker = makeReference(DataDistributionTrackerInitParams{ + .db = self->txnProcessor, + .distributorId = self->ddId, + .readyToStart = self->initialized, + .output = self->relocationProducer, + .shardsAffectedByTeamFailure = self->shardsAffectedByTeamFailure, + .physicalShardCollection = self->physicalShardCollection, + .bulkLoadTaskCollection = self->bulkLoadTaskCollection, + .anyZeroHealthyTeams = anyZeroHealthyTeams, + .shards = &shards, + .trackerCancelled = &self->context->trackerCancelled, + .usableRegions = self->configuration.usableRegions, + .shardEncodeLocationMetadata = self->context->ddEnabledState->shardEncodeLocationMetadata() }); actors.push_back(reportErrorsExcept(DataDistributionTracker::run(self->context->tracker, self->initData, getShardMetrics.getFuture(), diff --git a/fdbserver/workloads/CheckMetadataEncoding.cpp b/fdbserver/workloads/CheckMetadataEncoding.cpp index f9dc5637cb3..5210ad93273 100644 --- a/fdbserver/workloads/CheckMetadataEncoding.cpp +++ b/fdbserver/workloads/CheckMetadataEncoding.cpp @@ -127,7 +127,7 @@ struct CheckMetadataEncodingWorkload : TestWorkload { // matches the knob direction; a mismatched request is a test // misconfiguration and should fail loudly and immediately below. const bool pollForConvergence = (self->requireRollbackComplete && !self->shardEncodeExpected) || - (self->requireForwardComplete && self->shardEncodeExpected); + (self->requireForwardComplete && self->shardEncodeExpected); // Deadline is generous: under BUGGIFY the DD rollback rewrite can be // throttled hard (tiny SHARD_ENCODE_REWRITE_KS_BATCH_SIZE and // KRM_GET_RANGE_LIMIT), making Phase 2/3 legitimately take a few @@ -258,8 +258,7 @@ struct CheckMetadataEncodingWorkload : TestWorkload { // requested in principle; require whichever are set. bool terminalReached = true; if (self->requireRollbackComplete) { - terminalReached = - terminalReached && (keyServersNew == 0 && serverKeysNew == 0 && dataMovesCount == 0); + terminalReached = terminalReached && (keyServersNew == 0 && serverKeysNew == 0 && dataMovesCount == 0); } if (self->requireForwardComplete) { terminalReached = terminalReached && (keyServersOld == 0 && serverKeysOld == 0);