Repository: PolicyEngine/policyengine-core

Issue:
Fix atomic file write issue on windows
Failing on windows with permission error as windows does not support separate journal based file handler like unix systems

Discussion:

Solution patch:
diff --git a/.github/workflows/pr.yaml b/.github/workflows/pr.yaml
index 64cfaf51..b8cef923 100644
--- a/.github/workflows/pr.yaml
+++ b/.github/workflows/pr.yaml
@@ -31,7 +31,11 @@ jobs:
       - name: Check version number has been properly updated
         run: .github/is-version-number-acceptable.sh
   Test:
-    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        os: [ ubuntu-latest, windows-latest ]
+      fail-fast: false
+    runs-on: ${{ matrix.os }}
     steps:
       - name: Checkout repo
         uses: actions/checkout@v3
diff --git a/changelog_entry.yaml b/changelog_entry.yaml
index e69de29b..2ad2b9a8 100644
--- a/changelog_entry.yaml
+++ b/changelog_entry.yaml
@@ -0,0 +1,4 @@
+- bump: patch
+  changes:
+    fixed:
+    - Fix file writing on windows.
\ No newline at end of file
diff --git a/docs/add_plotly_to_book.py b/docs/add_plotly_to_book.py
index 822e77ab..05d49cef 100644
--- a/docs/add_plotly_to_book.py
+++ b/docs/add_plotly_to_book.py
@@ -14,7 +14,7 @@ args = parser.parse_args()
 book_folder = Path(args.book_path)
 
 for html_file in book_folder.glob("**/*.html"):
-    with open(html_file, "r") as f:
+    with open(html_file, "r", encoding="utf-8") as f:
         html = f.read()
 
     # Add the script tag to the start of the <head> tag.
@@ -23,5 +23,5 @@ for html_file in book_folder.glob("**/*.html"):
         '<head><script src="https://cdnjs.cloudflare.com/ajax/libs/require.js/2.3.4/require.min.js"></script>',
     )
 
-    with open(html_file, "w") as f:
+    with open(html_file, "w", encoding="utf-8") as f:
         f.write(html)
diff --git a/policyengine_core/data/dataset.py b/policyengine_core/data/dataset.py
index c6cec09f..d97852be 100644
--- a/policyengine_core/data/dataset.py
+++ b/policyengine_core/data/dataset.py
@@ -9,6 +9,7 @@ import os
 import tempfile
 from policyengine_core.tools.hugging_face import *
 import sys
+from policyengine_core.tools.win_file_manager import WindowsAtomicFileManager
 
 
 def atomic_write(file: Path, content: bytes) -> None:
@@ -22,20 +23,24 @@ def atomic_write(file: Path, content: bytes) -> None:
     If a process is reading the original file when a new file is renamed, that should relink the file, not clear and overwrite the old one so
     both processes should continue happily.
     """
-    with tempfile.NamedTemporaryFile(
-        mode="wb",
-        dir=file.parent.absolute().as_posix(),
-        prefix=file.name + ".download.",
-        delete=False,
-    ) as f:
-        try:
-            f.write(content)
-            f.close()
-            os.rename(f.name, file.absolute().as_posix())
-        except:
-            f.delete = True
-            f.close()
-            raise
+    if sys.platform == "win32":
+        manager = WindowsAtomicFileManager(file)
+        manager.write(content)
+    else:
+        with tempfile.NamedTemporaryFile(
+            mode="wb",
+            dir=file.parent.absolute().as_posix(),
+            prefix=file.name + ".download.",
+            delete=False,
+        ) as f:
+            try:
+                f.write(content)
+                f.close()
+                os.rename(f.name, file.absolute().as_posix())
+            except:
+                f.delete = True
+                f.close()
+                raise
 
 
 class Dataset:
diff --git a/policyengine_core/tools/win_file_manager.py b/policyengine_core/tools/win_file_manager.py
new file mode 100644
index 00000000..9228dcd5
--- /dev/null
+++ b/policyengine_core/tools/win_file_manager.py
@@ -0,0 +1,46 @@
+import os
+import tempfile
+from pathlib import Path
+from threading import Lock
+
+
+class WindowsAtomicFileManager:
+    """
+    https://stackoverflow.com/a/2368286
+    - Each instance manages a specific logical file by name.
+    - Files are written atomically using temporary files.
+    - Thread-safe operations are ensured using a `Lock`.
+    - Temporary files are created in the system's temporary directory.
+    - Replace the target file with the temporary file
+    - For any fallback, Cleanup temporary file if replacement fails
+    """
+
+    def __init__(self, file: Path):
+        self.logical_name = file.name
+        self.target_path = file
+        self.lock = Lock()
+
+    def write(self, content: bytes):
+        with self.lock:
+            with tempfile.NamedTemporaryFile(
+                mode="wb",
+                dir=self.target_path.parent.absolute().as_posix(),
+                delete=False,
+            ) as temp_file:
+                temp_file.write(content)
+                temp_file.flush()
+                os.fsync(temp_file.fileno())
+                temp_file_path = Path(temp_file.name)
+
+            try:
+                os.replace(temp_file_path, self.target_path)
+            except Exception as e:
+                temp_file_path.unlink(missing_ok=True)
+                raise e
+
+    def read(self) -> str:
+        with self.lock:
+            if not self.target_path.exists():
+                raise FileNotFoundError(f"{self.target_path} does not exist.")
+            with open(self.target_path, "rb") as file:
+                return file.read()


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 1da05584..6ab38421 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,5 +1,5 @@
 """
-This module directs PyTest to include certain global fixtures when running tests. 
+This module directs PyTest to include certain global fixtures when running tests.
 """
 
 pytest_plugins = [
diff --git a/tests/core/data/test_dataset.py b/tests/core/data/test_dataset.py
index f34e8316..cd166a37 100644
--- a/tests/core/data/test_dataset.py
+++ b/tests/core/data/test_dataset.py
@@ -1,5 +1,11 @@
 from pathlib import Path
 from tempfile import NamedTemporaryFile
+import sys
+import threading
+from policyengine_core.tools.win_file_manager import WindowsAtomicFileManager
+import tempfile
+from pathlib import Path
+import uuid
 
 
 def test_dataset_class():
@@ -28,18 +34,65 @@ def test_dataset_class():
 
 
 def test_atomic_write():
-    from policyengine_core.data.dataset import atomic_write
+    if sys.platform != "win32":
+        from policyengine_core.data.dataset import atomic_write
+
+        with NamedTemporaryFile(mode="w") as file:
+            file.write("Hello, world\n")
+            file.flush()
+            # Open the file before overwriting
+            with open(file.name, "r") as file_original:
+
+                atomic_write(Path(file.name), "NOPE\n".encode())
+
+                # Open file descriptor still points to the old node
+                assert file_original.readline() == "Hello, world\n"
+                # But if I open it again it has the new content
+                with open(file.name, "r") as file_updated:
+                    assert file_updated.readline() == "NOPE\n"
+
+
+def test_atomic_write_windows():
+    if sys.platform == "win32":
+        temp_dir = Path(tempfile.gettempdir())
+        temp_files = [
+            temp_dir / f"tempfile_{uuid.uuid4().hex}.tmp" for _ in range(5)
+        ]
+
+        managers = [WindowsAtomicFileManager(path) for path in temp_files]
+
+        contents_list = [
+            [f"Content_{i}_{j}".encode() for j in range(5)] for i in range(5)
+        ]
+
+        check_results = [[] for _ in range(5)]
+
+        threads = []
+        for i in range(5):
+            thread = threading.Thread(
+                target=file_task,
+                args=(managers[i], contents_list[i], check_results[i]),
+            )
+            threads.append(thread)
+            thread.start()
+
+        for thread in threads:
+            thread.join()
+
+        for i, results in enumerate(check_results):
+            for expected, actual in results:
+                assert (
+                    expected == actual
+                ), f"Mismatch in file {i}: expected {expected}, got {actual}"
 
-    with NamedTemporaryFile(mode="w") as file:
-        file.write("Hello, world\n")
-        file.flush()
-        # Open the file before overwriting
-        with open(file.name, "r") as file_original:
+        for temp_file in temp_files:
+            if temp_file.exists():
+                temp_file.unlink()
 
-            atomic_write(Path(file.name), "NOPE\n".encode())
 
-            # Open file descriptor still points to the old node
-            assert file_original.readline() == "Hello, world\n"
-            # But if I open it again it has the new content
-            with open(file.name, "r") as file_updated:
-                assert file_updated.readline() == "NOPE\n"
+def file_task(manager, contents, check_results):
+    for content in contents:
+        manager.write(content)
+        actual_content = manager.read().decode()
+        expected_content = content.decode()
+        check_results.append((expected_content, actual_content))