fakecloud_dynamodb/
export_import.rs

1//! Startup bulk-importer for AWS-format DynamoDB S3 exports (DYNAMODB_JSON).
2//!
3//! Reads a local copy of an AWS DynamoDB S3 export (manifests + gzipped data
4//! files) plus the table's `describe-table` JSON, and materialises it as a
5//! fresh table in state. The AWS attribute wire format IS our storage format
6//! (`AttributeValue` is a `serde_json::Value`), so items copy through verbatim
7//! with no type decoding.
8
9use std::collections::HashMap;
10use std::io::Read as _;
11use std::path::Path;
12
13use serde_json::Value;
14
15use crate::state::{DynamoTable, ProvisionedThroughput, SharedDynamoDbState};
16
17/// A single DynamoDB item: attribute name -> typed AWS wire value.
18type Item = HashMap<String, Value>;
19
20/// Outcome of a startup export import.
21///
22/// The import is idempotent: in persistent-storage mode the snapshot is loaded
23/// before this import runs, so on a restart with the same import flags the
24/// target table is already present. Rather than erroring (which would refuse to
25/// boot), the import is skipped and the already-loaded table is left untouched.
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub enum ImportOutcome {
28    /// A fresh table was materialised from the export with `items` items.
29    Imported { table: String, items: usize },
30    /// A table of the same name already existed in state; import was skipped and
31    /// the existing data was left untouched.
32    SkippedExisting { table: String },
33}
34
35/// Import an AWS-format DynamoDB export as a new table.
36///
37/// Idempotent: if the target table already exists in state (for example loaded
38/// from a persisted snapshot on restart), the import is skipped and the existing
39/// data is preserved instead of erroring out.
40pub fn import_aws_export(
41    state: &SharedDynamoDbState,
42    account_id: &str,
43    region: &str,
44    export_dir: &Path,      // folder holding manifest-summary.json etc.
45    describe_table: &Value, // parsed `aws dynamodb describe-table` JSON
46) -> Result<ImportOutcome, String> {
47    // Accept either the full `{"Table": {...}}` dump or a bare table object.
48    let shape = describe_table.get("Table").unwrap_or(describe_table);
49    let table_name = shape["TableName"]
50        .as_str()
51        .ok_or("describe-table: TableName missing or not a string")?
52        .to_string();
53
54    // Idempotent restart guard: if the table is already present (e.g. restored
55    // from a persisted snapshot before this import runs), leave it untouched and
56    // skip rather than error and refuse to boot. Checked before reading the
57    // export so a no-op restart does no work.
58    if table_exists(state, account_id, &table_name) {
59        tracing::warn!(
60            table = %table_name,
61            "skipping DynamoDB export import: table already exists in state; existing data left untouched"
62        );
63        return Ok(ImportOutcome::SkippedExisting { table: table_name });
64    }
65
66    let (items, declared_total) = read_export_items(export_dir)?;
67    let item_count = items.len();
68    // Manifest integrity: the summary's declared grand total must match what was
69    // actually read. Absent field -> skip the check (older/partial manifests).
70    if let Some(declared) = declared_total {
71        if declared != item_count {
72            return Err(format!(
73                "table {table_name}: manifest-summary declares itemCount {declared} but {item_count} items were read (truncated or corrupt export)"
74            ));
75        }
76    }
77    let table = build_table(&table_name, region, account_id, shape, items)?;
78
79    let mut guard = state.write();
80    let account = guard.get_or_create(account_id);
81    // Re-check under the write lock to close any check-then-insert gap and to
82    // handle a genuinely conflicting table gracefully (warn + skip, never panic).
83    if account.tables.contains_key(&table_name) {
84        tracing::warn!(
85            table = %table_name,
86            "skipping DynamoDB export import: table already exists in state; existing data left untouched"
87        );
88        return Ok(ImportOutcome::SkippedExisting { table: table_name });
89    }
90    account.tables.insert(table_name.clone(), table);
91
92    Ok(ImportOutcome::Imported {
93        table: table_name,
94        items: item_count,
95    })
96}
97
98/// True if `account_id` already holds a table named `table_name`.
99fn table_exists(state: &SharedDynamoDbState, account_id: &str, table_name: &str) -> bool {
100    state
101        .read()
102        .get(account_id)
103        .is_some_and(|account| account.tables.contains_key(table_name))
104}
105
106/// Walk the manifests (not a prefix listing) and read every item. Returns the
107/// items plus the summary manifest's declared `itemCount` (if present) so the
108/// caller can verify the grand total.
109fn read_export_items(export_dir: &Path) -> Result<(Vec<Item>, Option<usize>), String> {
110    let summary = read_json(&export_dir.join("manifest-summary.json"))?;
111    let format = summary["exportFormat"].as_str().unwrap_or_default();
112    if format != "DYNAMODB_JSON" {
113        return Err(format!(
114            "unsupported exportFormat {format:?} (only DYNAMODB_JSON is supported)"
115        ));
116    }
117
118    // `manifest-files.json` is JSON Lines (one object per line), each naming a
119    // data file via `dataFileS3Key`.
120    let manifest = read_text(&export_dir.join("manifest-files.json"))?;
121    let mut items = Vec::new();
122    for line in nonempty_lines(&manifest) {
123        let entry: Value =
124            serde_json::from_str(line).map_err(|e| format!("parse manifest-files line: {e}"))?;
125        let s3_key = entry["dataFileS3Key"]
126            .as_str()
127            .ok_or("manifest-files: dataFileS3Key missing")?;
128        // The manifest key uses the standard `AWSDynamoDB/{export-id}/data/<file>`
129        // layout; resolve it against the local export as `<export_dir>/data/<basename>`.
130        let basename = s3_key.rsplit('/').next().unwrap_or(s3_key);
131        let file_items = read_data_file(&export_dir.join("data").join(basename))?;
132        // Per-file integrity: when the manifest entry declares an itemCount it
133        // must match the records actually read from that data file.
134        if let Some(declared) = entry["itemCount"].as_u64() {
135            if declared as usize != file_items.len() {
136                return Err(format!(
137                    "data file {basename}: manifest declares itemCount {declared} but {} records were read (truncated or corrupt export)",
138                    file_items.len()
139                ));
140            }
141        }
142        items.extend(file_items);
143    }
144    let declared_total = summary["itemCount"].as_u64().map(|n| n as usize);
145    Ok((items, declared_total))
146}
147
148/// Read one gzipped data file; each line is a `{"Item": {...}}` record.
149fn read_data_file(path: &Path) -> Result<Vec<Item>, String> {
150    let file = std::fs::File::open(path).map_err(|e| format!("open {}: {e}", path.display()))?;
151    let mut contents = String::new();
152    flate2::read::GzDecoder::new(file)
153        .read_to_string(&mut contents)
154        .map_err(|e| format!("gunzip {}: {e}", path.display()))?;
155
156    // No dedup: AWS exports carry unique primary keys, so a single data file
157    // never contains a duplicate. Colliding keys would both be retained.
158    nonempty_lines(&contents)
159        .map(|line| {
160            let record: Value =
161                serde_json::from_str(line).map_err(|e| format!("parse data line: {e}"))?;
162            record["Item"]
163                .as_object()
164                .map(|obj| obj.clone().into_iter().collect())
165                .ok_or_else(|| "data line missing Item object".to_string())
166        })
167        .collect()
168}
169
170/// Build a fresh `DynamoTable` from a `describe-table` shape and its items.
171fn build_table(
172    name: &str,
173    region: &str,
174    account_id: &str,
175    shape: &Value,
176    items: Vec<Item>,
177) -> Result<DynamoTable, String> {
178    let key_schema =
179        crate::parse_key_schema(&shape["KeySchema"]).map_err(|e| format!("KeySchema: {e}"))?;
180    let attribute_definitions = crate::parse_attribute_definitions(&shape["AttributeDefinitions"])
181        .map_err(|e| format!("AttributeDefinitions: {e}"))?;
182
183    // Guard against a wrong/hand-edited describe-table producing an
184    // ACTIVE-but-key-corrupt table. For each declared key attribute (HASH and,
185    // if present, RANGE) every item must (1) carry the attribute and (2) carry
186    // it with the scalar type declared in AttributeDefinitions -- the same
187    // presence + type parity the normal write path enforces
188    // (validate_key_in_item / check_key_type). A wrong-typed key would let a
189    // correctly-typed read fail to find the row, so the data would appear to
190    // vanish. Fail the whole import so no partial table lands.
191    for elem in &key_schema {
192        if elem.key_type != "HASH" && elem.key_type != "RANGE" {
193            continue;
194        }
195        let role = if elem.key_type == "HASH" {
196            "partition"
197        } else {
198            "sort"
199        };
200        let expected_type = attribute_definitions
201            .iter()
202            .find(|d| d.attribute_name == elem.attribute_name)
203            .map(|d| d.attribute_type.as_str());
204        for item in &items {
205            let Some(value) = item.get(&elem.attribute_name) else {
206                return Err(format!(
207                    "table {name}: item missing {role} key attribute {:?} declared in KeySchema: {item:?}",
208                    elem.attribute_name
209                ));
210            };
211            if let Some(expected) = expected_type {
212                let actual = crate::state::attribute_type_and_value(value).map(|(ty, _)| ty);
213                if actual != Some(expected) {
214                    return Err(format!(
215                        "table {name}: {role} key attribute {:?} has type {} but AttributeDefinitions declares {expected}: {item:?}",
216                        elem.attribute_name,
217                        actual.unwrap_or("<none>"),
218                    ));
219                }
220            }
221        }
222    }
223
224    // BillingModeSummary wins over a bare BillingMode; default to PROVISIONED.
225    let billing_mode = shape["BillingModeSummary"]["BillingMode"]
226        .as_str()
227        .or_else(|| shape["BillingMode"].as_str())
228        .unwrap_or("PROVISIONED")
229        .to_string();
230    let provisioned_throughput = if billing_mode == "PAY_PER_REQUEST" {
231        ProvisionedThroughput {
232            read_capacity_units: 0,
233            write_capacity_units: 0,
234        }
235    } else {
236        crate::parse_provisioned_throughput(&shape["ProvisionedThroughput"])
237            .map_err(|e| format!("ProvisionedThroughput: {e}"))?
238    };
239
240    let mut table = DynamoTable {
241        name: name.to_string(),
242        arn: format!("arn:aws:dynamodb:{region}:{account_id}:table/{name}"),
243        table_id: uuid::Uuid::new_v4().to_string().replace('-', ""),
244        key_schema,
245        attribute_definitions,
246        provisioned_throughput,
247        items,
248        gsi: crate::parse_gsi(&shape["GlobalSecondaryIndexes"], &billing_mode),
249        lsi: crate::parse_lsi(&shape["LocalSecondaryIndexes"]),
250        tags: crate::parse_tags(&shape["Tags"]),
251        created_at: chrono::Utc::now(),
252        status: "ACTIVE".to_string(),
253        item_count: 0,
254        size_bytes: 0,
255        billing_mode,
256        ttl_attribute: None,
257        ttl_enabled: false,
258        resource_policy: None,
259        pitr_enabled: false,
260        kinesis_destinations: Vec::new(),
261        contributor_insights_status: "DISABLED".to_string(),
262        contributor_insights_counters: std::collections::BTreeMap::new(),
263        stream_enabled: false,
264        stream_view_type: None,
265        stream_arn: None,
266        stream_records: std::sync::Arc::new(parking_lot::RwLock::new(Vec::new())),
267        sse_type: None,
268        sse_kms_key_arn: None,
269        deletion_protection_enabled: false,
270        on_demand_throughput: None,
271        table_class: "STANDARD".to_string(),
272    };
273    table.recalculate_stats(); // fills item_count / size_bytes
274    Ok(table)
275}
276
277fn read_text(path: &Path) -> Result<String, String> {
278    std::fs::read_to_string(path).map_err(|e| format!("read {}: {e}", path.display()))
279}
280
281fn read_json(path: &Path) -> Result<Value, String> {
282    serde_json::from_str(&read_text(path)?).map_err(|e| format!("parse {}: {e}", path.display()))
283}
284
285fn nonempty_lines(text: &str) -> impl Iterator<Item = &str> {
286    text.lines().filter(|l| !l.trim().is_empty())
287}
288
289#[cfg(test)]
290mod tests {
291    use super::*;
292    use serde_json::json;
293    use std::path::PathBuf;
294
295    #[test]
296    fn imports_aws_export_into_state() {
297        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
298            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
299        ));
300
301        let fixtures = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
302        let describe = read_json(&fixtures.join("describe-table.json")).unwrap();
303        let export_dir = fixtures.join("export");
304
305        let outcome =
306            import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
307                .expect("import should succeed");
308        assert_eq!(
309            outcome,
310            ImportOutcome::Imported {
311                table: "Music".to_string(),
312                items: 3,
313            }
314        );
315
316        // Table lives in state with the right item count.
317        let mut guard = state.write();
318        let table = guard
319            .get_or_create("123456789012")
320            .tables
321            .get("Music")
322            .expect("table in state");
323        assert_eq!(table.items.len(), 3);
324        assert_eq!(table.item_count, 3);
325
326        // Type preservation: the "Foo" item carries every DynamoDB attribute
327        // type and each one survives verbatim. Complex types are asserted for
328        // FULL equality (not spot-checked) so a serializer regression that
329        // dropped a later list element / set member would be caught.
330        let foo = table
331            .items
332            .iter()
333            .find(|i| i.get("Artist") == Some(&json!({"S": "Foo"})))
334            .expect("item with Artist=Foo");
335
336        // Scalars: whole typed value.
337        assert_eq!(foo.get("str"), Some(&json!({"S": "hello"})));
338        assert_eq!(foo.get("Plays"), Some(&json!({"N": "42"})));
339        assert_eq!(foo.get("bin"), Some(&json!({"B": "aGVsbG8="})));
340        assert_eq!(foo.get("flag"), Some(&json!({"BOOL": true})));
341        assert_eq!(foo.get("nothing"), Some(&json!({"NULL": true})));
342
343        // List (3 mixed elements) and map (2 fields): exact whole-value equality.
344        assert_eq!(
345            foo.get("list"),
346            Some(&json!({"L": [{"S": "nested"}, {"N": "7"}, {"BOOL": false}]}))
347        );
348        assert_eq!(
349            foo.get("map"),
350            Some(&json!({"M": {"genre": {"S": "rock"}, "year": {"N": "1994"}}}))
351        );
352
353        // Sets: assert length AND every member (order is preserved verbatim,
354        // but check membership so a dropped-element regression can't hide).
355        let ss = foo.get("strset").unwrap()["SS"].as_array().unwrap();
356        assert_eq!(ss.len(), 2);
357        assert!(ss.contains(&json!("x")));
358        assert!(ss.contains(&json!("y")));
359
360        let ns = foo.get("numset").unwrap()["NS"].as_array().unwrap();
361        assert_eq!(ns.len(), 2);
362        assert!(ns.contains(&json!("1")));
363        assert!(ns.contains(&json!("2")));
364
365        let bs = foo.get("binset").unwrap()["BS"].as_array().unwrap();
366        assert_eq!(bs.len(), 2);
367        assert!(bs.contains(&json!("YQ==")));
368        assert!(bs.contains(&json!("Yg==")));
369    }
370
371    #[tokio::test]
372    async fn imported_table_persists_through_snapshot_save() {
373        // Regression: a startup bulk-import mutates state directly, so unless it
374        // is written through the DynamoDB snapshot store it is durable only if a
375        // later mutating API call happens to trigger a save. A read-only workload
376        // would lose the imported table on restart. Exercise import -> save ->
377        // reload (the exact path the server now wires) and assert the table
378        // survives.
379        use crate::save_dynamodb_snapshot;
380        use crate::state::DynamoDbSnapshot;
381        use fakecloud_persistence::{DiskSnapshotStore, SnapshotStore};
382        use std::sync::Arc;
383
384        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
385            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
386        ));
387
388        let fixtures = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
389        let describe = read_json(&fixtures.join("describe-table.json")).unwrap();
390        let export_dir = fixtures.join("export");
391        let outcome =
392            import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
393                .expect("import should succeed");
394        assert!(matches!(outcome, ImportOutcome::Imported { .. }));
395
396        let tmp = tempfile::TempDir::new().unwrap();
397        let store: Arc<dyn SnapshotStore> = Arc::new(DiskSnapshotStore::new(
398            tmp.path().join("dynamodb").join("snapshot.json"),
399        ));
400        let lock = tokio::sync::Mutex::new(());
401        let wrote = save_dynamodb_snapshot(&state, Some(store.clone()), &lock)
402            .await
403            .expect("snapshot save should succeed");
404        assert!(
405            wrote,
406            "a snapshot store is configured, so a save must occur"
407        );
408
409        // Reload exactly as boot does: read the bytes back and deserialize.
410        let bytes = store
411            .load()
412            .expect("load ok")
413            .expect("snapshot bytes on disk");
414        let snapshot: DynamoDbSnapshot =
415            serde_json::from_slice(&bytes).expect("snapshot deserializes");
416        let accounts = snapshot
417            .accounts
418            .expect("v2 multi-account snapshot written");
419        let account = accounts
420            .get("123456789012")
421            .expect("account present after reload");
422        let table = account
423            .tables
424            .get("Music")
425            .expect("imported table survives snapshot round-trip");
426        assert_eq!(table.items.len(), 3, "all imported items are durable");
427    }
428
429    #[test]
430    fn rejects_items_missing_declared_key_attribute() {
431        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
432            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
433        ));
434
435        let export_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/export");
436
437        // KeySchema names a partition key that the exported items don't carry;
438        // the import must fail rather than materialise a key-corrupt table.
439        let describe = json!({
440            "Table": {
441                "TableName": "Music",
442                "KeySchema": [{ "AttributeName": "DoesNotExist", "KeyType": "HASH" }],
443                "AttributeDefinitions": [
444                    { "AttributeName": "DoesNotExist", "AttributeType": "S" }
445                ],
446                "BillingMode": "PAY_PER_REQUEST"
447            }
448        });
449
450        let err = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
451            .expect_err("import should fail when items lack the declared key attribute");
452        assert!(
453            err.contains("DoesNotExist"),
454            "error should name the missing key attribute: {err}"
455        );
456    }
457
458    #[test]
459    fn rejects_key_attribute_type_mismatch() {
460        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
461            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
462        ));
463
464        let export_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/export");
465
466        // The exported items carry Artist as a string (`{"S": ...}`), but this
467        // describe-table declares the partition key as numeric (`N`). The import
468        // must fail with a type mismatch, matching the normal write path's
469        // check_key_type, rather than materialise a table a correctly-typed read
470        // could never query.
471        let describe = json!({
472            "Table": {
473                "TableName": "MusicTypeMismatch",
474                "KeySchema": [
475                    { "AttributeName": "Artist", "KeyType": "HASH" },
476                    { "AttributeName": "SongTitle", "KeyType": "RANGE" }
477                ],
478                "AttributeDefinitions": [
479                    { "AttributeName": "Artist", "AttributeType": "N" },
480                    { "AttributeName": "SongTitle", "AttributeType": "S" }
481                ],
482                "BillingMode": "PAY_PER_REQUEST"
483            }
484        });
485
486        let err = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
487            .expect_err("import should fail when a key attribute has the wrong type");
488        assert!(
489            err.contains("Artist") && err.contains("type S") && err.contains("declares N"),
490            "error should describe the key type mismatch: {err}"
491        );
492        // Nothing partial landed in state.
493        let mut guard = state.write();
494        assert!(!guard
495            .get_or_create("123456789012")
496            .tables
497            .contains_key("MusicTypeMismatch"));
498    }
499
500    #[test]
501    fn import_is_idempotent_when_table_already_exists() {
502        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
503            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
504        ));
505
506        let fixtures = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures");
507        let describe = read_json(&fixtures.join("describe-table.json")).unwrap();
508        let export_dir = fixtures.join("export");
509
510        // First import materialises the table (3 items).
511        let first = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
512            .expect("first import should succeed");
513        assert_eq!(
514            first,
515            ImportOutcome::Imported {
516                table: "Music".to_string(),
517                items: 3,
518            }
519        );
520
521        // Replace the loaded data with a single sentinel row, standing in for a
522        // table restored from a persisted snapshot that has since diverged.
523        {
524            let mut guard = state.write();
525            let table = guard
526                .get_or_create("123456789012")
527                .tables
528                .get_mut("Music")
529                .expect("table in state");
530            let mut sentinel: Item = HashMap::new();
531            sentinel.insert("Artist".to_string(), json!({ "S": "SENTINEL" }));
532            sentinel.insert("SongTitle".to_string(), json!({ "S": "only" }));
533            table.items = vec![sentinel];
534            table.recalculate_stats();
535        }
536
537        // Re-running with the same flags must skip (not error, not overwrite).
538        let second = import_aws_export(&state, "123456789012", "us-east-1", &export_dir, &describe)
539            .expect("re-import should be a no-op, not an error");
540        assert_eq!(
541            second,
542            ImportOutcome::SkippedExisting {
543                table: "Music".to_string(),
544            }
545        );
546
547        // Existing data was left untouched: still the single sentinel row, not
548        // re-populated with the 3 export items.
549        let mut guard = state.write();
550        let table = guard
551            .get_or_create("123456789012")
552            .tables
553            .get("Music")
554            .expect("table still in state");
555        assert_eq!(table.items.len(), 1);
556        assert_eq!(
557            table.items[0].get("Artist"),
558            Some(&json!({ "S": "SENTINEL" }))
559        );
560    }
561
562    #[test]
563    fn rejects_manifest_item_count_mismatch() {
564        use flate2::write::GzEncoder;
565        use flate2::Compression;
566        use std::io::Write as _;
567
568        // Build a throwaway export whose manifest-summary over-declares the item
569        // count (99) relative to the single row actually present, simulating a
570        // truncated/corrupt export.
571        let dir = std::env::temp_dir().join(format!("fc-ddb-import-{}", uuid::Uuid::new_v4()));
572        std::fs::create_dir_all(dir.join("data")).unwrap();
573        std::fs::write(
574            dir.join("manifest-summary.json"),
575            r#"{"version":"2020-06-30","exportFormat":"DYNAMODB_JSON","itemCount":99}"#,
576        )
577        .unwrap();
578        // Per-file entry carries no itemCount, so only the summary-level check
579        // can fire here.
580        std::fs::write(
581            dir.join("manifest-files.json"),
582            r#"{"dataFileS3Key":"AWSDynamoDB/01700000000000-abcd/data/0001.json.gz"}"#,
583        )
584        .unwrap();
585        let data_file = std::fs::File::create(dir.join("data/0001.json.gz")).unwrap();
586        let mut enc = GzEncoder::new(data_file, Compression::default());
587        enc.write_all(b"{\"Item\":{\"Artist\":{\"S\":\"A\"},\"SongTitle\":{\"S\":\"B\"}}}\n")
588            .unwrap();
589        enc.finish().unwrap();
590
591        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
592            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
593        ));
594        let describe = json!({
595            "Table": {
596                "TableName": "MusicTruncated",
597                "KeySchema": [
598                    { "AttributeName": "Artist", "KeyType": "HASH" },
599                    { "AttributeName": "SongTitle", "KeyType": "RANGE" }
600                ],
601                "AttributeDefinitions": [
602                    { "AttributeName": "Artist", "AttributeType": "S" },
603                    { "AttributeName": "SongTitle", "AttributeType": "S" }
604                ],
605                "BillingMode": "PAY_PER_REQUEST"
606            }
607        });
608
609        let err = import_aws_export(&state, "123456789012", "us-east-1", &dir, &describe)
610            .expect_err("import should fail when manifest itemCount disagrees with the data");
611        assert!(
612            err.contains("99") && err.contains("1 items"),
613            "error should report the declared vs actual counts: {err}"
614        );
615
616        let _ = std::fs::remove_dir_all(&dir);
617    }
618
619    #[test]
620    fn rejects_per_file_item_count_mismatch() {
621        use flate2::write::GzEncoder;
622        use flate2::Compression;
623        use std::io::Write as _;
624
625        // manifest-summary agrees with the data (1 item), but the per-file
626        // manifest entry over-declares itemCount (5). The per-file integrity
627        // check must catch it.
628        let dir = std::env::temp_dir().join(format!("fc-ddb-import-{}", uuid::Uuid::new_v4()));
629        std::fs::create_dir_all(dir.join("data")).unwrap();
630        std::fs::write(
631            dir.join("manifest-summary.json"),
632            r#"{"version":"2020-06-30","exportFormat":"DYNAMODB_JSON","itemCount":1}"#,
633        )
634        .unwrap();
635        std::fs::write(
636            dir.join("manifest-files.json"),
637            r#"{"itemCount":5,"dataFileS3Key":"AWSDynamoDB/01700000000000-abcd/data/0001.json.gz"}"#,
638        )
639        .unwrap();
640        let data_file = std::fs::File::create(dir.join("data/0001.json.gz")).unwrap();
641        let mut enc = GzEncoder::new(data_file, Compression::default());
642        enc.write_all(b"{\"Item\":{\"Artist\":{\"S\":\"A\"},\"SongTitle\":{\"S\":\"B\"}}}\n")
643            .unwrap();
644        enc.finish().unwrap();
645
646        let state: SharedDynamoDbState = std::sync::Arc::new(parking_lot::RwLock::new(
647            fakecloud_core::multi_account::MultiAccountState::new("123456789012", "us-east-1", ""),
648        ));
649        let describe = json!({
650            "Table": {
651                "TableName": "MusicPerFile",
652                "KeySchema": [
653                    { "AttributeName": "Artist", "KeyType": "HASH" },
654                    { "AttributeName": "SongTitle", "KeyType": "RANGE" }
655                ],
656                "AttributeDefinitions": [
657                    { "AttributeName": "Artist", "AttributeType": "S" },
658                    { "AttributeName": "SongTitle", "AttributeType": "S" }
659                ],
660                "BillingMode": "PAY_PER_REQUEST"
661            }
662        });
663
664        let err = import_aws_export(&state, "123456789012", "us-east-1", &dir, &describe)
665            .expect_err("import should fail when a data file's itemCount is wrong");
666        assert!(
667            err.contains("0001.json.gz") && err.contains('5'),
668            "error should name the data file and declared count: {err}"
669        );
670
671        let _ = std::fs::remove_dir_all(&dir);
672    }
673}
fakecloud_dynamodb/export_import.rs

fakecloud_dynamodb/
export_import.rs
