Skip to content

fix: write config files atomically #8793

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zouyonghe wants to merge 1 commit into
base: master
Choose a base branch
from
+58 −6
Open

fix: write config files atomically #8793

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 26 additions & 6 deletions astrbot/core/config/astrbot_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
import json
import logging
import os
import tempfile

from astrbot.core.utils.astrbot_path import get_astrbot_data_path
from astrbot.core.utils.auth_password import (
Expand Down Expand Up @@ -53,9 +54,9 @@ def __init__(

if not self.check_exist():
"""不存在时载入默认配置"""
with open(config_path, "w", encoding="utf-8-sig") as f:
json.dump(default_config, f, indent=4, ensure_ascii=False)
object.__setattr__(self, "first_deploy", True) # 标记第一次部署
self.update(default_config)
self.save_config(indent=4)
object.__setattr__(self, "first_deploy", True) # 标记第一次部署

with open(config_path, encoding="utf-8-sig") as f:
conf_str = f.read()
Expand Down Expand Up @@ -211,15 +212,34 @@ def check_config_integrity(self, refer_conf: dict, conf: dict, path=""):

return has_new

def save_config(self, replace_config: dict | None = None) -> None:
def save_config(

@sourcery-ai sourcery-ai Bot Jun 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (complexity): Consider extracting the atomic JSON write logic into a small helper so save_config stays a simple wrapper that focuses only on updating and delegating the write.

You can keep the atomic write + indentation behavior while pushing the low-level details out of save_config into a small helper. That keeps save_config at its old “orchestrator” level and makes the IO/cleanup logic easier to reason about and reuse.

For example:

import json
import os
import tempfile

def _atomic_write_json(path: str, data: dict, *, indent: int) -> None:
    directory = os.path.dirname(os.path.abspath(path)) or "."
    fd, temp_path = tempfile.mkstemp(
        dir=directory,
        prefix=f".{os.path.basename(path)}.",
        suffix=".tmp",
        text=True,
    )
    try:
        with os.fdopen(fd, "w", encoding="utf-8-sig") as f:
            json.dump(data, f, indent=indent, ensure_ascii=False)
            f.flush()
            os.fsync(f.fileno())
        os.replace(temp_path, path)
    except Exception:
        try:
            os.unlink(temp_path)
        except FileNotFoundError:
            pass
        raise

Then save_config becomes a thin wrapper again:

def save_config(
    self, replace_config: dict | None = None, *, indent: int = 2
) -> None:
    if replace_config:
        self.update(replace_config)
    _atomic_write_json(self.config_path, self, indent=indent)

This keeps all current behavior (atomic-ish write, custom indent, fsync, cleanup on error) but:

  • Removes nested IO/cleanup details from save_config
  • Makes the “save config” responsibility clearer and easier to scan
  • Gives you a reusable atomic JSON writer if you need it elsewhere.

self, replace_config: dict | None = None, *, indent: int = 2
) -> None:
"""将配置写入文件

如果传入 replace_config，则将配置替换为 replace_config
"""
if replace_config:
self.update(replace_config)
Comment on lines 222 to 223

@sourcery-ai sourcery-ai Bot Jun 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue: Using if replace_config will ignore valid but falsy configs such as an empty dict.

This means self.update is skipped when replace_config is an empty dict (or other falsy mapping), so you can’t clear the config via save_config. Use if replace_config is not None: to allow empty configs while still treating None as “no replacement provided.”

with open(self.config_path, "w", encoding="utf-8-sig") as f:
json.dump(self, f, indent=2, ensure_ascii=False)
directory = os.path.dirname(os.path.abspath(self.config_path)) or "."
fd, temp_path = tempfile.mkstemp(
dir=directory,
prefix=f".{os.path.basename(self.config_path)}.",
suffix=".tmp",
text=True,
)
Comment on lines +225 to +230

@gemini-code-assist gemini-code-assist Bot Jun 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Using text=True in tempfile.mkstemp opens the file descriptor in text mode (using O_TEXT on Windows). When you subsequently wrap this file descriptor with os.fdopen(fd, "w", encoding="utf-8-sig"), Python's io.TextIOWrapper expects a binary file descriptor to correctly handle the specified encoding and newline translation. Wrapping a text-mode file descriptor can lead to encoding conflicts, double-translation of line endings, or write corruption on Windows. Removing text=True (or setting it to False) ensures the file descriptor is opened in binary mode, allowing os.fdopen to portably and safely handle the UTF-8-SIG encoding.

        fd, temp_path = tempfile.mkstemp(
            dir=directory,
            prefix=f".{os.path.basename(self.config_path)}.",
            suffix=".tmp",
        )

try:
with os.fdopen(fd, "w", encoding="utf-8-sig") as f:
json.dump(self, f, indent=indent, ensure_ascii=False)
f.flush()
os.fsync(f.fileno())
os.replace(temp_path, self.config_path)
except Exception:
try:
os.unlink(temp_path)
except FileNotFoundError:
pass
raise

def __getattr__(self, item):
try:
Expand Down
32 changes: 32 additions & 0 deletions tests/unit/test_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -529,6 +529,38 @@ def test_save_config_with_replace(self, temp_config_path, minimal_default_config
# Original fields are preserved because update merges
assert "platform_settings" in loaded_config

def test_save_config_preserves_existing_file_when_write_fails(
self, temp_config_path, minimal_default_config, monkeypatch
):
"""Config saves should not corrupt the existing file on write failure."""
config = AstrBotConfig(
config_path=temp_config_path, default_config=minimal_default_config
)
with open(temp_config_path, encoding="utf-8-sig") as f:
original_content = f.read()

def failing_dump(*args, **kwargs):
file_obj = args[1]
file_obj.write("{")
raise RuntimeError("simulated interrupted write")

config.new_field = "new_value"
monkeypatch.setattr(
"astrbot.core.config.astrbot_config.json.dump",
failing_dump,
)

with pytest.raises(RuntimeError, match="simulated interrupted write"):
config.save_config()

with open(temp_config_path, encoding="utf-8-sig") as f:
assert f.read() == original_content
assert [
entry.name
for entry in os.scandir(os.path.dirname(temp_config_path))
if entry.name != os.path.basename(temp_config_path)
] == []
Comment on lines +556 to +562

@sourcery-ai sourcery-ai Bot Jun 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (testing): Temp-file cleanup assertion may be too strict and brittle for directories with other legitimate files

This assertion currently requires the directory to contain only the config file, which makes the test brittle if other legitimate files (e.g., from fixtures or future tests) are present. Instead, assert specifically that no temp files created by save_config remain, e.g. by filtering on the temp-file naming pattern:

assert not [
    e.name
    for e in os.scandir(os.path.dirname(temp_config_path))
    if e.name.startswith(f".{os.path.basename(temp_config_path)}.")
    and e.name.endswith(".tmp")
]
Suggested change
with open(temp_config_path, encoding="utf-8-sig") as f:
assert f.read() == original_content
assert [
entry.name
for entry in os.scandir(os.path.dirname(temp_config_path))
if entry.name != os.path.basename(temp_config_path)
] == []
with open(temp_config_path, encoding="utf-8-sig") as f:
assert f.read() == original_content
assert not [
entry.name
for entry in os.scandir(os.path.dirname(temp_config_path))
if entry.name.startswith(f".{os.path.basename(temp_config_path)}.")
and entry.name.endswith(".tmp")
]


def test_modification_persists_after_reload(
self, temp_config_path, minimal_default_config
):
Expand Down
Loading