fix(datasync): add guards and exception safety to JsonZstdFileSource

2026-01-03 16:21:59 -05:00
parent 57a44e0f3a
commit 2629cb26e0
1 changed files with 22 additions and 5 deletions
@@ -32,12 +32,29 @@ public sealed class JsonZstdFileSource : IImportSource

    public Task<IDataReader> ReadDataAsync(CancellationToken cancellationToken = default)
    {
-        _fileStream = new FileStream(_filePath, FileMode.Open, FileAccess.Read, FileShare.Read,
-            bufferSize: 65536, useAsync: true);
-        _decompressionStream = new DecompressionStream(_fileStream);
-        _reader = new JsonStreamingDataReader(_decompressionStream, _schema);
+        if (_fileStream != null)
+            throw new InvalidOperationException("ReadDataAsync has already been called. Dispose and create a new source to read again.");

-        return Task.FromResult<IDataReader>(_reader);
+        try
+        {
+            _fileStream = new FileStream(_filePath, FileMode.Open, FileAccess.Read, FileShare.Read,
+                bufferSize: 65536, useAsync: true);
+            _decompressionStream = new DecompressionStream(_fileStream);
+            _reader = new JsonStreamingDataReader(_decompressionStream, _schema);
+
+            return Task.FromResult<IDataReader>(_reader);
+        }
+        catch
+        {
+            // Clean up on failure
+            _reader?.Dispose();
+            _decompressionStream?.Dispose();
+            _fileStream?.Dispose();
+            _reader = null;
+            _decompressionStream = null;
+            _fileStream = null;
+            throw;
+        }
    }

    public async ValueTask DisposeAsync()