fix(secure_storage): process file system operations one at a time on Windows

packages/secure_storage/amplify_secure_storage_dart/lib/src/utils/file_key_value_store.dart

@@ -1,9 +1,11 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
+import 'package:aws_common/aws_common.dart';
 import 'package:file/file.dart' as pkg_file;
 import 'package:file/local.dart' as local_file;
 import 'package:meta/meta.dart';
@@ -18,7 +20,7 @@ import 'package:path/path.dart' as pkg_path;
 // without bringing in flutter as a dependency to the tests.
 class FileKeyValueStore {
   /// {@macro amplify_secure_storage_dart.file_key_value_store}
-  const FileKeyValueStore({
+  FileKeyValueStore({
     required this.path,
     required this.fileName,
     this.fs = const local_file.LocalFileSystem(),
@@ -32,6 +34,10 @@ class FileKeyValueStore {
   /// The file will be created if it does not yet exist.
   final String fileName;
 
+  final TaskScheduler _scheduler = TaskScheduler();
+
+  final AWSLogger logger = AWSLogger('FileKeyValueStore');
+
   @visibleForTesting
   final pkg_file.FileSystem fs;
 
@@ -47,9 +53,11 @@ class FileKeyValueStore {
     required String key,
     required Object value,
   }) async {
-    final data = await readAll();
-    data[key] = value;
-    return writeAll(data);
+    return _scheduler.schedule(() async {
+      final data = await readAll();
+      data[key] = value;
+      return writeAll(data);
+    });
   }
 
   /// Overwrites the existing data.
@@ -67,38 +75,87 @@ class FileKeyValueStore {
   Future<Object?> readKey({
     required String key,
   }) async {
-    final data = await readAll();
-    return data[key];
+    return _scheduler.schedule(() async {
+      final data = await readAll();
+      return data[key];
+    });
   }
 
   /// Removes a single key from storage.
   Future<void> removeKey({
     required String key,
   }) async {
-    final data = await readAll();
-    data.remove(key);
-    await writeAll(data);
+    return _scheduler.schedule(() async {
+      final data = await readAll();
+      data.remove(key);
+      await writeAll(data);
+    });
   }
 
   /// Returns true if the key exists in storage
   Future<bool> containsKey({
     required String key,
   }) async {
-    final data = await readAll();
-    return data.containsKey(key);
+    return _scheduler.schedule(() async {
+      final data = await readAll();
+      return data.containsKey(key);
+    });
   }
 
   /// Reads all the key-value pairs from storage.
   Future<Map<String, Object>> readAll() async {
     if (await file.exists()) {
       final stringMap = await file.readAsString();
       if (stringMap.isNotEmpty) {
-        final Object? data = json.decode(stringMap);
-        if (data is Map) {
-          return data.cast<String, Object>();
+        try {
+          final Object? data = json.decode(stringMap);
+          if (data is Map) {
+            return data.cast<String, Object>();
+          }
+        } on FormatException catch (e) {
+          logger.error(
+            'Cannot read file. The file may be corrupted. '
+            'Clearing file contents. See error for more details. '
+            'Error: $e',
+          );
+          await writeAll({});
         }
       }
     }
     return <String, Object>{};
   }
 }
+
+/// A class for processing async tasks one at a time in the order that they are
+/// scheduled.
+class TaskScheduler {
+  final _taskQueue = <Task<dynamic>>[];
+  bool _isProcessing = false;
+  Future<T> schedule<T>(Future<T> Function() task) {
+    final completer = Completer<T>();
+    _taskQueue.add(Task(task, completer));
+    _processTasks();
+    return completer.future;
+  }
+
+  Future<void> _processTasks() async {
+    if (_isProcessing) return;
+    _isProcessing = true;
+    while (_taskQueue.isNotEmpty) {
+      final currentTask = _taskQueue.removeAt(0);
+      try {
+        final result = await currentTask.task();
+        currentTask.completer.complete(result);
+      } on Object catch (e, stackTrace) {
+        currentTask.completer.completeError(e, stackTrace);
+      }
+    }
+    _isProcessing = false;
+  }
+}
+
+class Task<T> {
+  Task(this.task, this.completer);
+  final Future<T> Function() task;
+  final Completer<T> completer;
+}

packages/secure_storage/amplify_secure_storage_test/test/file_key_value_store_test.dart

@@ -3,8 +3,9 @@
 
 @TestOn('vm')
 
+import 'dart:io';
+
 import 'package:amplify_secure_storage_dart/src/utils/file_key_value_store.dart';
-import 'package:file/memory.dart';
 import 'package:test/test.dart';
 
 void main() {
@@ -14,10 +15,13 @@ void main() {
       storage = FileKeyValueStore(
         path: 'path',
         fileName: 'file',
-        fs: MemoryFileSystem(),
       );
     });
 
+    tearDown(() async {
+      await storage.file.delete();
+    });
+
     test('readKey & writeKey', () async {
       // assert initial state is null
       final value1 = await storage.readKey(key: 'key');
@@ -81,5 +85,42 @@ void main() {
       final includesKey2 = await storage.containsKey(key: 'key2');
       expect(includesKey2, isFalse);
     });
+
+    test('parallel writes should occur in the order they are called', () async {
+      final items = List.generate(1000, ((i) => i));
+      final futures = items.map(
+        (i) async => storage.writeKey(key: 'key', value: i),
+      );
+      await Future.wait(futures);
+      final value = await storage.readKey(key: 'key');
+      expect(value, items.last);
+    });
+
+    // Reference: https://github.com/aws-amplify/amplify-flutter/issues/5190
+    test('parallel write/remove operations should not corrupt the file',
+        () async {
+      final items = List.generate(1000, ((i) => i));
+      final futures = items.map(
+        (i) async {
+          if (i % 5 == 1) {
+            await storage.removeKey(key: 'key_${i - 1}');
+          }
+          return storage.writeKey(key: 'key_$i', value: 'value_$i');
+        },
+      );
+      await Future.wait(futures);
+    });
+
+    test('File is cleared when corrupted and can be re-written to', () async {
+      await storage.writeKey(key: 'foo', value: 'value');
+      final value1 = await storage.readKey(key: 'foo');
+      expect(value1, 'value');
+      await storage.file.writeAsString('{invalid json}', mode: FileMode.append);
+      final value2 = await storage.readKey(key: 'foo');
+      expect(value2, null);
+      await storage.writeKey(key: 'foo', value: 'value');
+      final value3 = await storage.readKey(key: 'foo');
+      expect(value3, 'value');
+    });
   });
 }