forked from jason/echo
ver 1.2
This commit is contained in:
jason
@@ -14,11 +14,18 @@ existing one is no longer wrongly skipped), correct `::` heading-target handling
|
||||
frontmatter field patches, a read-back-confirmed advisory multi-writer lock, a
|
||||
one-call cold-start `load`, and scope show/set.
|
||||
|
||||
Network is keep-alive and connection-pooled (one persistent connection per thread,
|
||||
reused across requests), and `read_many` fans bulk GETs across a thread pool — so
|
||||
full-vault scripts (sweep, lint, recall rebuild) make a handful of warm round-trips
|
||||
instead of hundreds of fresh TLS handshakes, and no longer blow past tool timeouts.
|
||||
|
||||
Config (env overrides; defaults match the rest of the plugin):
|
||||
ECHO_BASE default https://echoapi.alwisp.com
|
||||
ECHO_KEY default the plugin bearer token (env overrides it)
|
||||
ECHO_VERIFY default 1 — read-back verify after a PUT
|
||||
ECHO_LOCK_TTL default 900 — seconds before an advisory lock is considered stale
|
||||
ECHO_TIMEOUT default 30 — per-request socket timeout (seconds)
|
||||
ECHO_WORKERS default 8 — concurrency for read_many bulk reads
|
||||
ECHO_TODAY YYYY-MM-DD to use as "today" (pass the conversation's currentDate)
|
||||
|
||||
Usage:
|
||||
@@ -46,15 +53,16 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as dt
|
||||
import http.client
|
||||
import json
|
||||
import locale
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from pathlib import Path
|
||||
|
||||
# Never let a non-ASCII byte (em-dash in our own output, or Unicode in a fetched
|
||||
@@ -76,6 +84,11 @@ DEFAULT_KEY = "241265fbe6830934a9a4ad3e69335f64a42153b663aa5b0017cb1ea1217b2bab"
|
||||
KEY = echo_secrets.resolve_key(DEFAULT_KEY)
|
||||
VERIFY = os.environ.get("ECHO_VERIFY", "1") != "0"
|
||||
LOCK_TTL = int(os.environ.get("ECHO_LOCK_TTL", "900"))
|
||||
# Per-request socket timeout. A single stuck file fails fast instead of stalling a
|
||||
# full-vault pass; with read_many concurrency it only blocks one worker, not the run.
|
||||
TIMEOUT = int(os.environ.get("ECHO_TIMEOUT", "30"))
|
||||
# Concurrency for bulk reads (read_many). I/O-bound, so threads bypass the GIL fine.
|
||||
MAX_WORKERS = max(1, int(os.environ.get("ECHO_WORKERS", "8")))
|
||||
|
||||
LOCK_PATH = "_agent/locks/vault.lock"
|
||||
|
||||
@@ -101,30 +114,105 @@ def vault_url(path: str) -> str:
|
||||
return f"{BASE}/vault/{safe}"
|
||||
|
||||
|
||||
# --- persistent connection pool (keep-alive) ---------------------------------
|
||||
# urllib.urlopen opened a fresh TCP+TLS connection per call; against a remote HTTPS
|
||||
# endpoint the repeated handshake dominated, so a full-vault pass made hundreds of
|
||||
# serial handshakes and blew past the tool/sandbox timeout. We keep ONE connection
|
||||
# per thread (http.client is not thread-safe, so each thread — main + every pool
|
||||
# worker — owns its own) and reuse it across requests. read_many fans out over a
|
||||
# thread pool, so the pool's connections stay warm for the duration of a sweep.
|
||||
_local = threading.local()
|
||||
MAX_ATTEMPTS = 3
|
||||
|
||||
|
||||
def _new_connection() -> http.client.HTTPConnection:
|
||||
parts = urllib.parse.urlsplit(BASE)
|
||||
host = parts.hostname or parts.path
|
||||
if parts.scheme == "http":
|
||||
return http.client.HTTPConnection(host, parts.port or 80, timeout=TIMEOUT)
|
||||
return http.client.HTTPSConnection(host, parts.port or 443, timeout=TIMEOUT)
|
||||
|
||||
|
||||
def _connection() -> http.client.HTTPConnection:
|
||||
conn = getattr(_local, "conn", None)
|
||||
if conn is None:
|
||||
conn = _new_connection()
|
||||
_local.conn = conn
|
||||
return conn
|
||||
|
||||
|
||||
def _drop_connection() -> None:
|
||||
conn = getattr(_local, "conn", None)
|
||||
if conn is not None:
|
||||
try:
|
||||
conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
_local.conn = None
|
||||
|
||||
|
||||
def request(method: str, url: str, data: bytes | None = None,
|
||||
headers: dict[str, str] | None = None) -> tuple[int, bytes]:
|
||||
"""One bounded retry on transport failure (status 0) or 5xx. Returns (status, body)."""
|
||||
"""Issue one request over this thread's reused connection. Returns (status, body);
|
||||
status 0 == transport failure (echo_queue relies on this to trigger offline queueing).
|
||||
Retries: a stale pooled connection reconnects immediately; 5xx and other transport
|
||||
errors get one bounded sleep-retry, same as before."""
|
||||
merged = {"Authorization": f"Bearer {KEY}", **(headers or {})}
|
||||
parts = urllib.parse.urlsplit(url)
|
||||
target = parts.path or "/"
|
||||
if parts.query:
|
||||
target += "?" + parts.query
|
||||
last_status, last_body = 0, b""
|
||||
for attempt in range(2):
|
||||
req = urllib.request.Request(url, data=data, method=method, headers=merged)
|
||||
for attempt in range(MAX_ATTEMPTS):
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as response:
|
||||
return response.status, response.read()
|
||||
except urllib.error.HTTPError as exc:
|
||||
body = exc.read()
|
||||
if exc.code >= 500 and attempt == 0:
|
||||
conn = _connection()
|
||||
conn.request(method, target, body=data, headers=merged)
|
||||
resp = conn.getresponse()
|
||||
body = resp.read() # drain fully so the connection can be reused
|
||||
status = resp.status
|
||||
if status >= 500 and attempt < MAX_ATTEMPTS - 1:
|
||||
_drop_connection()
|
||||
time.sleep(1)
|
||||
continue
|
||||
return exc.code, body
|
||||
except Exception as exc: # connection error, timeout, DNS, ...
|
||||
return status, body
|
||||
except (http.client.RemoteDisconnected, http.client.BadStatusLine,
|
||||
ConnectionResetError, BrokenPipeError) as exc:
|
||||
# The server closed an idle keep-alive connection between requests.
|
||||
# Expected, not a failure: reconnect and retry without backing off.
|
||||
_drop_connection()
|
||||
last_status, last_body = 0, str(exc).encode()
|
||||
if attempt == 0:
|
||||
continue
|
||||
except Exception as exc: # timeout, DNS, refused, TLS, ...
|
||||
_drop_connection()
|
||||
last_status, last_body = 0, str(exc).encode()
|
||||
if attempt < MAX_ATTEMPTS - 1:
|
||||
time.sleep(1)
|
||||
continue
|
||||
return last_status, last_body
|
||||
|
||||
|
||||
def get_text(path: str) -> str | None:
|
||||
"""GET a vault path -> decoded text, or None on 404/any error. Never raises, so a
|
||||
single unreadable note can't abort a bulk pass (resilience for sweep/lint)."""
|
||||
try:
|
||||
status, body = request("GET", vault_url(path))
|
||||
except Exception:
|
||||
return None
|
||||
return body.decode(errors="replace") if status == 200 else None
|
||||
|
||||
|
||||
def read_many(paths) -> dict[str, str | None]:
|
||||
"""Concurrently GET many vault paths -> {path: text_or_None}. The single biggest
|
||||
win for full-vault scripts: it turns hundreds of serial round-trips into MAX_WORKERS
|
||||
parallel ones over warm keep-alive connections. Resilient — a bad file maps to None."""
|
||||
unique = list(dict.fromkeys(paths))
|
||||
if not unique:
|
||||
return {}
|
||||
workers = min(MAX_WORKERS, len(unique))
|
||||
with ThreadPoolExecutor(max_workers=workers) as pool:
|
||||
return dict(zip(unique, pool.map(get_text, unique)))
|
||||
|
||||
|
||||
def check(status: int, body: bytes, context: str) -> None:
|
||||
if status == 404:
|
||||
raise EchoError(f"{context}: not found", 44)
|
||||
|
||||
Reference in New Issue
Block a user