fix: manifest duplication - #1749
Conversation
issue occurred due to improper updating during writing manifests. Each standalone node now inserts in correct file without duplication
WalkthroughThe changes persist standalone node identity, use it for manifest naming and ownership, repair duplicate manifest entries during migration, and make staging metadata scans tolerant of invalid candidates. ChangesManifest identity and repair
Estimated code review effort: 4 (Complex) | ~45 minutes Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
|
@coderabbitai review |
✅ Action performedReview finished.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/catalog/mod.rs (1)
357-371: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winA missing manifest object still appends a duplicate snapshot entry.
posidentifies an entry this writer owns. Ifget_manifestreturnsNone, for example because the manifest object was deleted while the snapshot entry survived, this branch callscreate_manifestwithupdate_snapshot = false. That returnsSome(new_snapshot_entry), andfinalize_snapshot_updateappends it tometa.snapshot.manifest_list. The new entry carries the manifest path derived from the same stream, date, and writer, so it equalsmanifests[pos].manifest_path. The list then holds two entries with an identical path, and the condition repeats on every sync cycle.Remove or replace the stale entry at
posbefore creating the replacement manifest.🐛 Proposed fix
} else { - // Manifest not found, create new one - create_manifest( + // Manifest object is gone; drop the stale snapshot entry so the replacement + // entry does not duplicate it. + manifests.remove(pos); + create_manifest( partition_lower, partition_changes, stream_name, false, meta.clone(),Note that
meta.clone()must happen after the removal for the cloned metadata to stay consistent; adjust the borrow order as needed.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/catalog/mod.rs` around lines 357 - 371, In the missing-manifest branch around create_manifest, remove or replace the stale manifest entry at pos before creating the replacement. Perform this mutation before calling meta.clone(), then pass the updated metadata into create_manifest while preserving the existing creation flow and snapshot finalization behavior.
🧹 Nitpick comments (2)
src/catalog/mod.rs (1)
265-276: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winMatch the manifest file name exactly rather than by substring.
containsaccepts any stored path in which the writer name appears. A node id that is a suffix of another writer name, for example id01ABCagainst a stored path.../pod-01ABC.manifest.json, matches wrongly. Use the file-name component instead.♻️ Proposed change
- let manifest_file_name = manifest_path("").to_string(); + let manifest_file_name = manifest_path("") + .file_name() + .unwrap_or_default() + .to_string(); let pos = meta.snapshot.manifest_list.iter().position(|item| { item.time_lower_bound <= partition_lower && partition_lower < item.time_upper_bound - && item.manifest_path.contains(&manifest_file_name) + && item + .manifest_path + .rsplit('/') + .next() + .is_some_and(|name| name == manifest_file_name) });🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/catalog/mod.rs` around lines 265 - 276, Update the ownership predicate in the manifest_list position lookup to compare the stored path’s file-name component exactly with manifest_file_name, replacing the substring contains check. Preserve the existing time-overlap condition and ensure paths with writer names that merely contain the target name do not match.src/handlers/http/modal/server.rs (1)
151-159: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winPropagate all metadata-load errors instead of panicking.
Use
get_or_try_init(...).await?. Also replace bothput_on_disk(...).expect(...)calls in the metadata loader with?; disk failures there still abort the process.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/handlers/http/modal/server.rs` around lines 151 - 159, Update the NODE_META initialization to use get_or_try_init(...).await? so metadata-loading errors propagate instead of panicking, and make the metadata loader propagate both put_on_disk failures with ? rather than expect(...).
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/handlers/http/modal/mod.rs`:
- Around line 439-456: Update the metadata scan loop around Self::from_bytes to
avoid returning the first valid candidate from nondeterministic read_dir order.
Collect all successfully parsed metadata candidates, select one
deterministically using sorted file names, and preserve the existing read and
parse error logging; optionally warn when multiple valid candidates are found.
In `@src/migration/mod.rs`:
- Around line 446-455: In the unrecognized-version branch surrounding
duplicates_removed, avoid converting stream_metadata_value through
ObjectStoreFormat before persistence. Either skip put_stream_json entirely for
unknown versions, or persist the repaired raw serde_json::Value directly so
newer fields are preserved and conversion errors cannot abort migration.
---
Outside diff comments:
In `@src/catalog/mod.rs`:
- Around line 357-371: In the missing-manifest branch around create_manifest,
remove or replace the stale manifest entry at pos before creating the
replacement. Perform this mutation before calling meta.clone(), then pass the
updated metadata into create_manifest while preserving the existing creation
flow and snapshot finalization behavior.
---
Nitpick comments:
In `@src/catalog/mod.rs`:
- Around line 265-276: Update the ownership predicate in the manifest_list
position lookup to compare the stored path’s file-name component exactly with
manifest_file_name, replacing the substring contains check. Preserve the
existing time-overlap condition and ensure paths with writer names that merely
contain the target name do not match.
In `@src/handlers/http/modal/server.rs`:
- Around line 151-159: Update the NODE_META initialization to use
get_or_try_init(...).await? so metadata-loading errors propagate instead of
panicking, and make the metadata loader propagate both put_on_disk failures with
? rather than expect(...).
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: ae939292-9cd4-416b-94ef-8805ed53eb2a
📒 Files selected for processing (6)
src/catalog/mod.rssrc/handlers/http/modal/mod.rssrc/handlers/http/modal/server.rssrc/migration/mod.rssrc/migration/stream_metadata_migration.rssrc/storage/object_storage.rs
| let bytes = match std::fs::read(&path) { | ||
| Ok(bytes) => bytes, | ||
| Err(e) => { | ||
| error!("Couldn't read {}: {}", path.display(), e); | ||
| continue; | ||
| } | ||
| }; | ||
| match Self::from_bytes(&bytes, options.flight_port) { | ||
| Ok(meta) => return Some(meta), | ||
| Err(e) => { | ||
| error!("Failed to extract {} metadata: {}", node_type_str, e); | ||
| return None; | ||
| continue; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| None | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
Scan result depends on directory order when several metadata files exist.
The loop returns the first parsable candidate. read_dir order is not defined. If staging holds more than one {node_type}.{id}.json file, for example after an earlier run wrote a different id, the resolved node identity can change between restarts. That defeats the stable-identity goal of this PR and can re-introduce manifest duplication.
Consider collecting candidates and selecting deterministically, for example by sorted file name, or log a warning when more than one valid candidate is present.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/handlers/http/modal/mod.rs` around lines 439 - 456, Update the metadata
scan loop around Self::from_bytes to avoid returning the first valid candidate
from nondeterministic read_dir order. Collect all successfully parsed metadata
candidates, select one deterministically using sorted file names, and preserve
the existing read and parse error logging; optionally warn when multiple valid
candidates are found.
| // If the version is not recognized, we assume it's already in the latest format. | ||
| // The snapshot repair above still needs persisting when it changed anything. | ||
| if duplicates_removed > 0 { | ||
| let stream_json: ObjectStoreFormat = | ||
| serde_json::from_value(stream_metadata_value.clone())?; | ||
| PARSEABLE | ||
| .metastore | ||
| .put_stream_json(&stream_json, stream, tenant_id) | ||
| .await?; | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Persisting an unrecognized version through ObjectStoreFormat can drop fields.
This branch handles versions the binary does not know, which includes a stream.json written by a newer build. serde_json::from_value::<ObjectStoreFormat> keeps only the fields this build declares, so put_stream_json writes back a document without any newer keys. The conversion can also fail outright and abort migration for the stream, where the previous code returned the value unchanged.
The repair is idempotent and already applied in memory, so persisting here is optional. Either skip the write for unknown versions, or write the repaired Value without the typed round trip.
🐛 Proposed change
_ => {
- // If the version is not recognized, we assume it's already in the latest format.
- // The snapshot repair above still needs persisting when it changed anything.
- if duplicates_removed > 0 {
- let stream_json: ObjectStoreFormat =
- serde_json::from_value(stream_metadata_value.clone())?;
- PARSEABLE
- .metastore
- .put_stream_json(&stream_json, stream, tenant_id)
- .await?;
- }
+ // The version is not recognized, so the document may come from a newer build.
+ // A typed round trip would drop fields this build does not know, and the repair
+ // is idempotent, so the in-memory result is returned without persisting.
return Ok(stream_metadata_value);
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // If the version is not recognized, we assume it's already in the latest format. | |
| // The snapshot repair above still needs persisting when it changed anything. | |
| if duplicates_removed > 0 { | |
| let stream_json: ObjectStoreFormat = | |
| serde_json::from_value(stream_metadata_value.clone())?; | |
| PARSEABLE | |
| .metastore | |
| .put_stream_json(&stream_json, stream, tenant_id) | |
| .await?; | |
| } | |
| // The version is not recognized, so the document may come from a newer build. | |
| // A typed round trip would drop fields this build does not know, and the repair | |
| // is idempotent, so the in-memory result is returned without persisting. | |
| return Ok(stream_metadata_value); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/migration/mod.rs` around lines 446 - 455, In the unrecognized-version
branch surrounding duplicates_removed, avoid converting stream_metadata_value
through ObjectStoreFormat before persistence. Either skip put_stream_json
entirely for unknown versions, or persist the repaired raw serde_json::Value
directly so newer fields are preserved and conversion errors cannot abort
migration.
1 participant
issue occurred due to improper updating during writing manifests. Each standalone node now inserts in correct file without duplication
Fixes #XXXX.
Description
This PR has:
Summary by CodeRabbit