Harden stage-2 enrichment: durable sidecar, enforceable timeout, tests
Address code-review findings on the metadata pipeline: - finalize_known! now fsyncs the sidecar bytes before the rename and fsyncs done/ after, so the "file in done/ implies sidecar present" invariant holds across power loss, not just process crashes. The docstring previously claimed an fsync the code never performed. - run_exiftool's timeout escalates SIGTERM -> (2s grace) -> SIGKILL, so an exiftool that ignores SIGTERM can't pin a worker forever on wait(proc). Previously the timeout sent only SIGTERM. - Add test/ (48 tests) covering the correctness-critical paths: sanitize_filename, normalize_metadata, degraded build_metadata, real exiftool extraction, finalize_known! end-to-end, recover_dir!.
This commit is contained in:
@@ -34,6 +34,24 @@ const AUTHOR_TAGS = ["Author", "Artist", "By-line", "Owner", "Artist"]
|
||||
const CREATED_DATE_TAGS = ["DateTimeOriginal", "CreateDate", "MediaCreateDate", "CreationDate"]
|
||||
const MODIFIED_DATE_TAGS = ["ModifyDate", "FileModifyDate"]
|
||||
|
||||
"fsync an open file descriptor, throwing on failure — used to make a write durable before a rename commits it."
|
||||
function fsync_fd(fd)
|
||||
ccall(:fsync, Cint, (Cint,), fd) == 0 || error("fsync failed: $(Base.Libc.strerror())")
|
||||
return nothing
|
||||
end
|
||||
|
||||
"fsync a directory so a rename into it survives a crash (the rename, not just the file bytes, must be persisted)."
|
||||
function fsync_dir(dir::AbstractString)
|
||||
dfd = ccall(:open, Cint, (Cstring, Cint), dir, 0) # O_RDONLY
|
||||
dfd < 0 && error("cannot open dir for fsync: $dir ($(Base.Libc.strerror()))")
|
||||
try
|
||||
fsync_fd(dfd)
|
||||
finally
|
||||
ccall(:close, Cint, (Cint,), dfd)
|
||||
end
|
||||
return nothing
|
||||
end
|
||||
|
||||
"Strip exiftool's `-G` group prefix (`EXIF:Software` → `Software`) so lookups are group-agnostic."
|
||||
strip_group(tag::AbstractString) = String(last(split(tag, ':')))
|
||||
|
||||
@@ -72,7 +90,14 @@ function run_exiftool(path::AbstractString, timeout::Integer)
|
||||
end
|
||||
if process_running(proc)
|
||||
killed[] = true
|
||||
kill(proc)
|
||||
kill(proc, Base.SIGTERM)
|
||||
# Escalate: a process that ignores/defers SIGTERM would otherwise pin
|
||||
# the worker forever on the wait(proc) below, defeating the timeout.
|
||||
grace = 0.0
|
||||
while process_running(proc) && grace < 2.0
|
||||
sleep(0.1); grace += 0.1
|
||||
end
|
||||
process_running(proc) && kill(proc, Base.SIGKILL)
|
||||
end
|
||||
end
|
||||
wait(proc)
|
||||
@@ -154,10 +179,12 @@ end
|
||||
Commit an enriched known file to `done/` with the sidecar-first ordering so the
|
||||
invariant *"a file in done/ implies its sidecar is already there"* always holds.
|
||||
|
||||
Sequence: write `<name>.meta.json` directly into `done/`, fsync-close it, THEN
|
||||
move the file into `done/`. A crash between the two leaves only a harmless orphan
|
||||
sidecar in `done/` while the file stays in `known/`, so stage-aware recovery
|
||||
re-drives it and overwrites the sidecar — idempotent.
|
||||
Sequence: write `<name>.meta.json` to a temp name, fsync its bytes, rename it
|
||||
into place, fsync `done/` so the rename itself is durable, THEN move the file
|
||||
into `done/`. A crash between the two leaves only a harmless orphan sidecar in
|
||||
`done/` while the file stays in `known/`, so stage-aware recovery re-drives it
|
||||
and overwrites the sidecar — idempotent. The fsyncs make the ordering hold
|
||||
across power loss, not just process crashes.
|
||||
"""
|
||||
function finalize_known!(cfg::Config, job::Job, meta)
|
||||
base = basename(job.path)
|
||||
@@ -168,8 +195,11 @@ function finalize_known!(cfg::Config, job::Job, meta)
|
||||
# sidecar and a crash mid-write can't masquerade as a committed one.
|
||||
open(tmp_sidecar, "w") do io
|
||||
write(io, JSON3.write(meta))
|
||||
flush(io)
|
||||
fsync_fd(fd(io)) # durably persist bytes before the rename
|
||||
end
|
||||
mv(tmp_sidecar, sidecar; force=true) # sidecar committed first
|
||||
fsync_dir(cfg.done_dir) # persist the rename itself, not just the bytes
|
||||
|
||||
file_dest = move_to(cfg.done_dir, job) # file arrival = commit point
|
||||
return (file_dest, sidecar)
|
||||
|
||||
Reference in New Issue
Block a user