341f154c36
deploy/stack.yaml passes DATABASE_URL_FILE, AWS_ACCESS_KEY_ID_FILE, OPERATOR_PASSWORD_FILE and the provider tokens as Swarm secret paths, but the runtime only ever read the plain names. That stack could not start: the database URL and R2 credentials were absent, and operator login raised KeyError, so it returned 500 instead of the intended 503. local/secrets.py resolves every <NAME>_FILE into <NAME> before configuration is read, from the API, worker and bootstrap entrypoints. It fails closed on an unreadable or empty secret and on a name supplied both directly and as a file, because starting with a credential nobody intended is worse than not starting. Only one trailing newline is stripped, so a generated password keeps any whitespace that belongs to it, and no value reaches an error message. The stack also passed OPERATOR_USER while the Kanban authenticates by email; it now passes OPERATOR_EMAIL, matching the runtime. The release gate checked this by searching local/secrets.py for the literal "DATABASE_URL_FILE", which would pass for any file containing that string. It now loads the module and makes it resolve every secret the stack declares, and asserts it fails closed on a missing one. Four marker strings that stopped matching when R2 support landed are removed rather than left to rot; the two that still describe real blockers stay, so the gate continues to refuse a release while payment and messaging adapters are fake. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
80 lines
3.5 KiB
Python
80 lines
3.5 KiB
Python
"""Transactional outbox worker. Fake receipts persist; no messages leave the stack."""
|
|
import json
|
|
import logging
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from psycopg.types.json import Jsonb
|
|
from .adapters import FakeTiny, FakeWhatsApp, LocalS3Storage, require_runtime
|
|
from .secrets import load as load_secret_files
|
|
from .db import connect
|
|
from .scanning import ClamAV, scan_loop
|
|
|
|
load_secret_files()
|
|
require_runtime()
|
|
adapters = {'tiny': FakeTiny(), 'whatsapp': FakeWhatsApp()}
|
|
last_tick = 0.0
|
|
last_cleanup = 0.0
|
|
storage = LocalS3Storage()
|
|
scan_thread = None
|
|
|
|
def cleanup():
|
|
"""Delete only expired object bytes; retain order/file metadata and history."""
|
|
with connect() as c:
|
|
rows = c.execute("""SELECT * FROM dtf_local.uploads WHERE purged_at IS NULL
|
|
AND (expires_at<=now() OR (NOT complete AND created_at<now()-interval '1 day'))
|
|
ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 50""").fetchall()
|
|
for row in rows:
|
|
storage.discard(row['object_key'],row['multipart_id'],row['complete'])
|
|
c.execute('UPDATE dtf_local.uploads SET purged_at=now() WHERE id=%s', (row['id'],))
|
|
c.execute('DELETE FROM dtf_local.sessions WHERE expires_at<=now()')
|
|
c.execute('DELETE FROM dtf_local.operator_sessions WHERE expires_at<=now()')
|
|
c.execute("DELETE FROM dtf_local.login_attempts WHERE started_at<now()-interval '1 day'")
|
|
c.execute("DELETE FROM dtf_local.security_events WHERE created_at<now()-interval '30 days'")
|
|
|
|
def tick():
|
|
global last_tick
|
|
with connect() as c:
|
|
job = c.execute('SELECT * FROM dtf_local.outbox WHERE delivered_at IS NULL AND available_at <= now() ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1').fetchone()
|
|
if job:
|
|
try:
|
|
receipt = adapters[job['provider']].deliver(job['event_key'], job['payload'])
|
|
c.execute('UPDATE dtf_local.outbox SET delivered_at=now(), receipt=%s, attempts=attempts+1, last_error=NULL WHERE id=%s', (Jsonb(receipt),job['id']))
|
|
except Exception as exc:
|
|
delay = min(1800, 2**min(job['attempts']+1, 10))
|
|
c.execute("UPDATE dtf_local.outbox SET attempts=attempts+1, last_error=%s, available_at=now() + %s * interval '1 second' WHERE id=%s", (str(exc),delay,job['id']))
|
|
last_tick = time.monotonic()
|
|
|
|
def loop():
|
|
global last_cleanup
|
|
while True:
|
|
try:
|
|
tick()
|
|
if time.monotonic()-last_cleanup > 60:
|
|
cleanup()
|
|
last_cleanup = time.monotonic()
|
|
except Exception:
|
|
logging.exception('Local worker tick failed')
|
|
time.sleep(1)
|
|
|
|
class Health(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
scanner = False
|
|
try:
|
|
scanner = bool(scan_thread and scan_thread.is_alive() and ClamAV().ping())
|
|
except Exception:
|
|
pass
|
|
healthy = time.monotonic()-last_tick < 15 and scanner
|
|
self.send_response(200 if self.path == '/health' and healthy else 503)
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps({'worker': 'ok' if healthy else 'unavailable',
|
|
'scanner': 'ok' if scanner else 'unavailable'}).encode())
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
if __name__ == '__main__':
|
|
scan_thread = threading.Thread(target=scan_loop,args=(storage,),daemon=True)
|
|
scan_thread.start()
|
|
threading.Thread(target=loop, daemon=True).start()
|
|
HTTPServer(('0.0.0.0',8002),Health).serve_forever()
|