fix(db): каталог данных 0700, одна копия при миграции под блокировкой (#13)
M-7 (hub#4), M-2 и M-10 (hub#12). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
70bcd4cf25
commit
ec49ed64ba
@ -2,10 +2,12 @@
|
|||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
from . import paths
|
||||||
|
|
||||||
|
|
||||||
def make_backup(conn, backups_dir: Path, label: str, keep: int = 30) -> Path:
|
def make_backup(conn, backups_dir: Path, label: str, keep: int = 30) -> Path:
|
||||||
backups_dir = Path(backups_dir)
|
backups_dir = Path(backups_dir)
|
||||||
backups_dir.mkdir(parents=True, exist_ok=True)
|
paths.secure_dir(backups_dir)
|
||||||
target = backups_dir / f"{datetime.now():%Y%m%d-%H%M%S-%f}-{label}.sqlite"
|
target = backups_dir / f"{datetime.now():%Y%m%d-%H%M%S-%f}-{label}.sqlite"
|
||||||
conn.execute("VACUUM INTO ?", (str(target),))
|
conn.execute("VACUUM INTO ?", (str(target),))
|
||||||
# копии перед миграцией схемы (db.migrate) — для отката, ротации не подлежат
|
# копии перед миграцией схемы (db.migrate) — для отката, ротации не подлежат
|
||||||
|
|||||||
@ -6,7 +6,7 @@ from contextlib import contextmanager
|
|||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from . import backup
|
from . import backup, paths
|
||||||
|
|
||||||
SCHEMA_VERSION = 2
|
SCHEMA_VERSION = 2
|
||||||
|
|
||||||
@ -118,7 +118,7 @@ def connect(path) -> sqlite3.Connection:
|
|||||||
операций — transaction()."""
|
операций — transaction()."""
|
||||||
is_memory = str(path) == ":memory:"
|
is_memory = str(path) == ":memory:"
|
||||||
if not is_memory:
|
if not is_memory:
|
||||||
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
paths.secure_dir(Path(path).parent)
|
||||||
conn = sqlite3.connect(str(path), timeout=10, isolation_level=None, check_same_thread=False)
|
conn = sqlite3.connect(str(path), timeout=10, isolation_level=None, check_same_thread=False)
|
||||||
conn.row_factory = sqlite3.Row
|
conn.row_factory = sqlite3.Row
|
||||||
if not is_memory:
|
if not is_memory:
|
||||||
@ -134,22 +134,29 @@ def _schema_version(conn) -> int | None:
|
|||||||
|
|
||||||
|
|
||||||
def migrate(conn: sqlite3.Connection, path=None) -> None:
|
def migrate(conn: sqlite3.Connection, path=None) -> None:
|
||||||
"""Создать недостающие таблицы и поднять схему существующей базы до SCHEMA_VERSION. Перед миграцией файловой базы —
|
"""Создать недостающие таблицы и поднять схему существующей базы до SCHEMA_VERSION. Копия базы
|
||||||
копия в <каталог базы>/backups (без ротации). Параллельный connect безопасен: версия перечитывается в транзакции."""
|
(<каталог базы>/backups, без ротации) и миграция — под одной блокировкой записи (BEGIN IMMEDIATE): параллельный
|
||||||
|
connect дождётся её, перечитает версию и не станет ни копировать, ни мигрировать повторно (M-2 #12). Копию делает
|
||||||
|
отдельное соединение: VACUUM INTO внутри транзакции SQLite запрещает, а читатель при WAL видит базу до миграции."""
|
||||||
conn.executescript(SCHEMA)
|
conn.executescript(SCHEMA)
|
||||||
version = _schema_version(conn)
|
version = _schema_version(conn)
|
||||||
if version is None:
|
if version is None:
|
||||||
conn.execute("INSERT INTO setting(key, value) VALUES ('schema_version', ?)", (json.dumps(SCHEMA_VERSION),))
|
conn.execute("INSERT OR IGNORE INTO setting(key, value) VALUES ('schema_version', ?)", (json.dumps(SCHEMA_VERSION),))
|
||||||
elif version < SCHEMA_VERSION:
|
elif version < SCHEMA_VERSION:
|
||||||
if path is not None:
|
|
||||||
backup.make_backup(conn, Path(path).parent / "backups", f"schema-v{version}", keep=0)
|
|
||||||
with transaction(conn):
|
with transaction(conn):
|
||||||
current = _schema_version(conn)
|
current = _schema_version(conn)
|
||||||
while current < SCHEMA_VERSION:
|
if current < SCHEMA_VERSION:
|
||||||
for sql in MIGRATIONS[current]:
|
if path is not None:
|
||||||
conn.execute(sql)
|
reader = sqlite3.connect(str(path), timeout=10)
|
||||||
current += 1
|
try:
|
||||||
conn.execute("UPDATE setting SET value = ? WHERE key = 'schema_version'", (json.dumps(current),))
|
backup.make_backup(reader, Path(path).parent / "backups", f"schema-v{current}", keep=0)
|
||||||
|
finally:
|
||||||
|
reader.close()
|
||||||
|
while current < SCHEMA_VERSION:
|
||||||
|
for sql in MIGRATIONS[current]:
|
||||||
|
conn.execute(sql)
|
||||||
|
current += 1
|
||||||
|
conn.execute("UPDATE setting SET value = ? WHERE key = 'schema_version'", (json.dumps(current),))
|
||||||
conn.execute("INSERT OR IGNORE INTO agent_status(id) VALUES (1)")
|
conn.execute("INSERT OR IGNORE INTO agent_status(id) VALUES (1)")
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@ -1,7 +1,10 @@
|
|||||||
"""Где лежат данные хаба: каталог, база, резервные копии."""
|
"""Где лежат данные хаба: каталог, база, резервные копии."""
|
||||||
|
import logging
|
||||||
import os
|
import os
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def home() -> Path:
|
def home() -> Path:
|
||||||
return Path(os.environ.get("JIRA_HUB_HOME", str(Path.home() / ".local" / "share" / "jira-hub")))
|
return Path(os.environ.get("JIRA_HUB_HOME", str(Path.home() / ".local" / "share" / "jira-hub")))
|
||||||
@ -13,3 +16,14 @@ def db_path() -> Path:
|
|||||||
|
|
||||||
def backups_dir() -> Path:
|
def backups_dir() -> Path:
|
||||||
return home() / "backups"
|
return home() / "backups"
|
||||||
|
|
||||||
|
|
||||||
|
def secure_dir(path: Path) -> None:
|
||||||
|
"""Создать каталог и закрыть его владельцу (0700): в базе и копиях лежат токены (M-7). Сбой chmod (каталог чужой) —
|
||||||
|
предупреждение, хаб продолжает работу."""
|
||||||
|
path = Path(path)
|
||||||
|
path.mkdir(parents=True, exist_ok=True)
|
||||||
|
try:
|
||||||
|
os.chmod(path, 0o700)
|
||||||
|
except OSError as e:
|
||||||
|
log.warning("каталог %s не закрыт (0700): %s", path, e)
|
||||||
|
|||||||
@ -1,6 +1,10 @@
|
|||||||
|
import logging
|
||||||
|
import os
|
||||||
import sqlite3
|
import sqlite3
|
||||||
|
import stat
|
||||||
|
import threading
|
||||||
|
|
||||||
from jira_hub import db
|
from jira_hub import backup, db, paths, queue, settings
|
||||||
|
|
||||||
|
|
||||||
def test_connect_creates_schema():
|
def test_connect_creates_schema():
|
||||||
@ -110,3 +114,67 @@ def test_nested_transaction_rollback_is_local(tmp_path):
|
|||||||
except RuntimeError:
|
except RuntimeError:
|
||||||
pass
|
pass
|
||||||
assert [r["title"] for r in conn.execute("SELECT title FROM batch")] == ["outer"]
|
assert [r["title"] for r in conn.execute("SELECT title FROM batch")] == ["outer"]
|
||||||
|
|
||||||
|
|
||||||
|
def _mode(p):
|
||||||
|
return stat.S_IMODE(p.stat().st_mode)
|
||||||
|
|
||||||
|
|
||||||
|
def test_data_dir_is_private(tmp_path):
|
||||||
|
home = tmp_path / "data"
|
||||||
|
home.mkdir()
|
||||||
|
home.chmod(0o755)
|
||||||
|
db.connect(home / "hub.sqlite").close()
|
||||||
|
assert _mode(home) == 0o700
|
||||||
|
|
||||||
|
|
||||||
|
def test_backups_dir_is_private(tmp_path):
|
||||||
|
conn = db.connect(tmp_path / "hub.sqlite")
|
||||||
|
backup.make_backup(conn, tmp_path / "backups", "batch-1")
|
||||||
|
conn.close()
|
||||||
|
assert _mode(tmp_path / "backups") == 0o700
|
||||||
|
|
||||||
|
|
||||||
|
def test_secure_dir_failure_only_warns(tmp_path, monkeypatch, caplog):
|
||||||
|
def deny(*a, **k):
|
||||||
|
raise PermissionError("нет прав")
|
||||||
|
monkeypatch.setattr(os, "chmod", deny)
|
||||||
|
with caplog.at_level(logging.WARNING):
|
||||||
|
paths.secure_dir(tmp_path / "x")
|
||||||
|
assert (tmp_path / "x").is_dir() and "0700" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
|
def test_parallel_connect_makes_one_backup(tmp_path):
|
||||||
|
path = tmp_path / "hub.sqlite"
|
||||||
|
_v1_database(path)
|
||||||
|
barrier, errors = threading.Barrier(4), []
|
||||||
|
|
||||||
|
def go():
|
||||||
|
try:
|
||||||
|
barrier.wait()
|
||||||
|
db.connect(path).close()
|
||||||
|
except Exception as e: # noqa: BLE001
|
||||||
|
errors.append(e)
|
||||||
|
|
||||||
|
threads = [threading.Thread(target=go) for _ in range(4)]
|
||||||
|
for t in threads:
|
||||||
|
t.start()
|
||||||
|
for t in threads:
|
||||||
|
t.join(30)
|
||||||
|
assert errors == []
|
||||||
|
assert len(list((tmp_path / "backups").glob("*-schema-v1.sqlite"))) == 1
|
||||||
|
conn = db.connect(path)
|
||||||
|
assert conn.execute("SELECT value FROM setting WHERE key = 'schema_version'").fetchone()["value"] == "2"
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
|
||||||
|
def test_migrated_claimed_redelivered_end_to_end(tmp_path):
|
||||||
|
"""База v1 с командой в claimed -> миграция (deliveries = 1) -> claim довыдаёт её с deliveries = 2 (M-10 #12)."""
|
||||||
|
path = tmp_path / "hub.sqlite"
|
||||||
|
_v1_database(path)
|
||||||
|
conn = db.connect(path)
|
||||||
|
settings.ensure_defaults(conn)
|
||||||
|
cmd = queue.claim(conn)
|
||||||
|
assert cmd["id"] == 1
|
||||||
|
assert queue.get_command(conn, 1)["deliveries"] == 2
|
||||||
|
conn.close()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user