fix: unique temp on JSON write + tolerant created_at render (v0.1.1)

- JsonStore._write used a fixed '<path>.tmp' name with no lock, so two concurrent
  authorizer invocations could clobber each other's temp and corrupt/lose the key
  store. use tempfile.mkstemp in the same dir (unique per write) then os.replace
  (atomic), cleaning up the temp on failure.
- list 'created_at' formatting did int(raw) unguarded; one hand-edited/legacy doc
  with a bad timestamp aborted the whole table. guard per-row, fall back to '-'.

verified by execution: 20 concurrent writers -> 0 errors, file stays valid JSON,
no leftover .tmp; upsert still dedupes/updates; bad/absent created_at -> '-'.

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
disqualifier 2026-06-28 15:47:58 -04:00
parent fb733e86de
commit a0824c4b1a
4 changed files with 30 additions and 10 deletions

View File

@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "envelope_authorizer"
version = "0.1.0"
version = "0.1.1"
description = "CLI key-authorization manager for envelope_crypto"
requires-python = ">=3.10"
dependencies = [

View File

@ -1 +1 @@
__version__ = "0.1.0"
__version__ = "0.1.1"

View File

@ -19,11 +19,16 @@ def _can_authorize(crypto, doc) -> str:
def _created(doc) -> str:
"""format created_at as a UTC timestamp string, or '-' if absent"""
"""format created_at as a UTC timestamp string, or '-' if absent/unparseable"""
raw = doc.get("meta", {}).get("created_at")
if not raw:
return "-"
return datetime.fromtimestamp(int(raw), tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
try:
return datetime.fromtimestamp(int(raw), tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
except (TypeError, ValueError, OverflowError, OSError):
# one hand-edited/legacy doc with a bad created_at shouldn't abort the
# whole table render
return "-"
def run(config, storage, args) -> None:

View File

@ -8,6 +8,7 @@ setups; for dev->server flows across machines, use the mongo backend.
import json
import os
import tempfile
from pathlib import Path
from typing import List, Optional
@ -34,13 +35,27 @@ class JsonStore(StorageBackend):
return data
def _write(self, docs: List[dict]) -> None:
"""atomically write the docs list (temp file then replace)"""
"""atomically write the docs list (unique temp file then replace)
the temp name is unique per write (tempfile.mkstemp in the same dir) so two
concurrent writers can't clobber a shared `.tmp`; os.replace is atomic on the
same filesystem. the temp is cleaned up if the write fails before replace.
"""
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_suffix(self.path.suffix + ".tmp")
with open(tmp, "w", encoding="utf-8") as handle:
json.dump(docs, handle, indent=2)
handle.write("\n")
os.replace(tmp, self.path)
fd, tmp = tempfile.mkstemp(
dir=self.path.parent, prefix=self.path.name + ".", suffix=".tmp"
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(docs, handle, indent=2)
handle.write("\n")
os.replace(tmp, self.path)
except BaseException:
try:
os.unlink(tmp)
except OSError:
pass
raise
def get(self, fingerprint: str) -> Optional[dict]:
"""return the doc whose `_id` matches, or None"""