Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
81e2e1cfed |
@@ -10,6 +10,8 @@ VEDASTRO_SELF_HOST_ENDPOINT=https://your-domain.example.com/vedastro
|
||||
# VEDASTRO_ENABLE_NETWORK=0
|
||||
|
||||
# Cache / TTL / free-tier queue. These keep ordinary users from calling upstream directly.
|
||||
# 失败结果另有短 TTL(VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS,默认 300 秒,且不超过成功 TTL)。
|
||||
# 这些缓存含出生资料,7 天为上限。
|
||||
VEDASTRO_CACHE_TTL_SECONDS=604800
|
||||
VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS=604800
|
||||
VEDASTRO_GATEWAY_QUEUE_ENABLED=1
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
# 印度占星 Skill 更新日志
|
||||
|
||||
## 2026-10-14 — 服务器不再长期保留出生资料副本
|
||||
|
||||
- 排盘缓存、官方快照和相关临时文件会按期限删掉,不再一直堆在服务器上。审计用的快照文件默认不再落地。失败的官方快照只短时间复用。
|
||||
- Skill 版本不变(BUG-1330~1332)。
|
||||
|
||||
## 2026-10-14 — 婚恋评级改成证据档位,KP 判宫和大运年长对齐(待合入 staging)
|
||||
|
||||
- 婚恋结构指数不再写成婚姻稳定、离婚、再婚或建议婚前咨询。中文报告用中文档位,英文报告用英文档位。配偶星分数按 0 到 1 显示,不再写成百分制。
|
||||
|
||||
@@ -17795,3 +17795,48 @@
|
||||
- 复发自:无。
|
||||
- 修复版本:未修复。
|
||||
|
||||
## BUG-1330 | 排盘缓存只进不出,同一张盘还会存成多份
|
||||
|
||||
- 状态:fixed-pending-verify
|
||||
- 首次发现 / 最近更新:2026-10-13 / 2026-10-14
|
||||
- 来源:issue #60。任务书 `docs/tasks/TASK-api-disk-cache-retention-20261013.md` T1、T3。
|
||||
- 现象:排盘结果写在磁盘缓存里。过期只在读取时判断,文件一直留着。启动时的清理扫的是异步任务目录,不是这份缓存。同一张盘因为小时、经纬度写成数字或字符串,会各存一份。
|
||||
- 触发条件:反复排盘,或同一次出生时间用不同写法提交。服务长时间不重启时,异步任务记录也会一直留到被轮询。
|
||||
- 根因:写缓存的人只做了读取时的过期判断。写清理的人以为缓存和异步任务在同一个目录、同一种记录。缓存键直接用了请求里的原值,计算前却会先把这些值解析成数字。
|
||||
- 修复:新的留存模块按缓存写入时间和排盘缓存期限删除过期文件,坏文件超过一天也删。启动时清理一次,之后按周期在后台再清。新写入的缓存文件改为仅本人可读,并先写临时文件再换上。缓存键把能解析的数字收成同一种写法,缓存格式版本从 5 改为 6,旧文件自然失效后被清掉。快速排盘和完整排盘仍然是两把不同的键。
|
||||
- 验证:`tests/test_api_scratch_retention.py`、`tests/test_api_async_job_contract.py`、`tests/test_api_server_security.py -k cache`、`tests/test_api_server_growth_contract.py`。数值写法的逐字节比较见 `scripts/research/api_chart_cache_byte_diff.py`。
|
||||
- 防复发:新增磁盘缓存必须同时在 `scripts/api_scratch_retention.py` 登记留存规则。`tests/test_api_scratch_retention.py` 会检查 `scripts/` 里新的 `scratch/local/` 目录有没有登记。
|
||||
- 相关记录:BUG-161、BUG-1331、BUG-1332。
|
||||
- 复发自:无。
|
||||
- 修复版本:分支 `codex/api-disk-cache-retention-20261013`,尚未合入 staging。
|
||||
|
||||
## BUG-1331 | VedAstro 审计文件留着出生资料,失败快照被缓存太久
|
||||
|
||||
- 状态:fixed-pending-verify
|
||||
- 首次发现 / 最近更新:2026-10-13 / 2026-10-14
|
||||
- 来源:issue #60。任务书 T2。
|
||||
- 现象:每次官方快照归一都会在磁盘留一份带缩进的审计文件,里面有出生时间。失败、不完整、被挡住的快照和成功结果用同一个长缓存期限。状态接口报出的期限没有算上回落值,可能和实际不一致。
|
||||
- 触发条件:在线排盘或报告走到官方快照。样例环境把成功缓存写成 7 天时,一次失败也会被原样复用 7 天。
|
||||
- 根因:审计落盘最早只给离线比对用,后来同一条归一进了在线请求,开关没有跟着关掉。语义缓存把「拿到一个结果」当成「可以长期复用」。状态接口自己读了一个变量,没有走实际生效的那个函数。
|
||||
- 修复:审计落盘默认关闭,需要比对时再显式打开。打开时改为紧凑文件、仅本人可读、先写临时文件再换上。失败、不完整、被挡住的快照用短期限,默认 300 秒,并且不会长过成功期限。状态接口改为报出含回落的成功期限,并单独报出失败期限。过期的语义缓存、业务响应缓存和审计文件由留存模块删除。
|
||||
- 验证:`tests/test_api_scratch_retention.py` 覆盖默认关闭、打开后的文件、失败 301 秒后不再命中、状态接口回落。既有 VedAstro 适配器测试仍通过。需要审计文件的那一条测试改为显式打开开关。
|
||||
- 防复发:新增磁盘缓存必须同时在 `scripts/api_scratch_retention.py` 登记留存规则。
|
||||
- 相关记录:BUG-734、BUG-301、BUG-1330。
|
||||
- 复发自:无。
|
||||
- 修复版本:分支 `codex/api-disk-cache-retention-20261013`,尚未合入 staging。
|
||||
|
||||
## BUG-1332 | 前台快照、负缓存、网关任务和分盘段缓存没有清理
|
||||
|
||||
- 状态:fixed-pending-verify
|
||||
- 首次发现 / 最近更新:2026-10-13 / 2026-10-14
|
||||
- 来源:issue #60。任务书 T4。
|
||||
- 现象:前台快照缓存超过回看窗口后不会再被读,但文件不删。负缓存只认当天,昨天的文件一直在。网关任务和分盘段缓存也只增不减。这些目录在持久卷上,部署不会清空。
|
||||
- 触发条件:使用前台官方快照、咨询负缓存、网关队列或校正分盘段。
|
||||
- 根因:当初只规定了什么时候还能读,没有规定什么时候可以删。
|
||||
- 修复:前台快照早于「今天减去回看天数再减一天」就删。负缓存早于今天(UTC)就删。已完成或失败、并且超过保留时间的网关任务连同原始响应一起删;排队中和进行中的不删。分盘段按修改时间保留,默认 7 天。只删符合原有文件名的文件,其它文件不碰。读写和缓存键都没有改。
|
||||
- 验证:`tests/test_api_scratch_retention.py` 覆盖这四个目录的留、删和不碰。`tests/test_vedastro_snapshot_cache.py` 与 `tests/test_vedastro_consultation_snapshot.py` 断言未改。
|
||||
- 防复发:新增磁盘缓存必须同时在 `scripts/api_scratch_retention.py` 登记留存规则。
|
||||
- 相关记录:BUG-727、BUG-1231、BUG-1330。
|
||||
- 复发自:无。
|
||||
- 修复版本:分支 `codex/api-disk-cache-retention-20261013`,尚未合入 staging。
|
||||
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
# PROGRESS · API 磁盘缓存留存(2026-10-14)
|
||||
|
||||
- 分支:`codex/api-disk-cache-retention-20261013`
|
||||
- 任务书基线写的是 `77057a819`。本分支从当时的 `origin/staging` `66cd60081f37a57680274633fcb3fb942d01bb2f` 拉出。A/B 的父提交用的是这个 SHA,不是任务书里的旧基线。
|
||||
- 接口文件行数:10946(验收口径 ≤ 10970)。类方法数和伪造点没有增加。
|
||||
|
||||
## 做了什么
|
||||
|
||||
- T1:新增 `scripts/api_scratch_retention.py`。启动时清理全部已登记目录,并按 `JYOTISH_SCRATCH_PRUNE_INTERVAL_SECONDS`(默认 1800,0 表示不启后台线程)周期清理。排盘缓存新文件改为仅本人可读,先写临时文件再换上。`prune_expired_async_jobs` 只留同名薄包装。
|
||||
- T2:`VEDASTRO_WRITE_AUDIT_ARTIFACTS` 默认关。关掉时不写审计文件,`artifact_path` 为 `None`。打开时写紧凑 JSON,仅本人可读,先写临时文件再换上。成功快照仍用原 TTL;失败、不完整、被挡住的用短 TTL,默认 300 秒,并且不超过成功 TTL。旧缓存没有 `negative` 标记时仍按成功的长期限读。状态接口改报含回落的成功期限,并多报失败期限。`.env.cn.example` 只加了注释,两个 604800 没改。
|
||||
- 成功的判据:`available is True` 且 `status == "ok"`。`partial` 即使 `available` 为真也算失败缓存。
|
||||
- T3:年月日能整除才收成整数,不截断 1990.5。时分秒和经纬度、时区能解析成有限数字就收成浮点,不四舍五入。解析不了的保持原值。日期取前 10 位,空值仍是空值。缓存格式版本 5 → 6。快速排盘和完整排盘仍是两把键。
|
||||
- T4:前台快照、负缓存、网关任务、分盘段都进了登记表。只删符合原文件名的过期文件。网关排队中和进行中的不删。读写逻辑和键没改。
|
||||
- T5:脚本在 `scripts/research/api_disk_cache_ab.py`,逐字节比较在 `scripts/research/api_chart_cache_byte_diff.py`。记录在 `docs/testing/api-disk-cache-retention-ab-20261013.md`。
|
||||
- T6:没有改咨询卡片缓存。数字用任务书 §2.5,没有重测:未命中构建 1.9 秒;文件 5097751 字节;命中时解析中位 68.5 毫秒;整次命中 82.5 毫秒;两天大约每个活跃用户 10 MB。可选方向仍是两条:按领域切片只读需要的部分;清理改为每隔几分钟一次,而不是每次读取都扫全目录。没有做进程内已解析包缓存。
|
||||
- T7:BUG-1330、BUG-1331、BUG-1332。变更说明一句。任务索引改为待验收。没有改界面说明。
|
||||
|
||||
## 命令和结果
|
||||
|
||||
| 检查 | 命令 | 结果 |
|
||||
| --- | --- | --- |
|
||||
| 新留存、异步任务、增长合同 | `pytest tests/test_api_scratch_retention.py tests/test_api_async_job_contract.py tests/test_api_server_growth_contract.py -q` | 26 通过 |
|
||||
| 安全缓存 | `pytest tests/test_api_server_security.py -q -k cache` | 6 通过 |
|
||||
| 任务书列出的 VedAstro / 咨询套件 | 见任务书 §8 那一条 | 4 失败,见下。其余通过 |
|
||||
| 需要审计文件的执行器测试 | `pytest tests/test_vedastro_service_adapter_executor.py::test_vedastro_range_scan_records_hashes_and_artifact_path -q` | 1 通过 |
|
||||
| 逐字节比较与 A/B | `python scripts/research/api_disk_cache_ab.py` | 退出码 0。命中率 B/A = 1.333。七天后 B 剩 40 个文件,等于当天键数。审计文件 0。负缓存剩 1。一万个文件的清理 20.174 秒 |
|
||||
| 接口行数 | 按换行计数 | 10946 |
|
||||
|
||||
既有断言改动只有一处,在 `tests/test_vedastro_service_adapter_executor.py`:原值是子进程沿用进程环境;新值是显式打开审计落盘;原因是默认关闭后这条测试仍要检查文件存在。这是测试前置,不是放宽断言。
|
||||
|
||||
## 未达到的目标(目标不改)
|
||||
|
||||
- 混合写法再重复 3 次时,命中率只是 A 的 1.333 倍,不是 2.5 倍。重复本身已经让 A 命中三分之二。
|
||||
- 一万个过期文件的清理用了 20.174 秒,不是 2 秒内。清理不在请求线程。默认间隔和失败短期限没有因此改掉。
|
||||
|
||||
## 本机套件里原先就会红的三条
|
||||
|
||||
- `tests/test_vedastro_gateway.py` 要求原始响应文件权限是 0600。Windows 报出来的是可写,不是 0600。这条断言没改。新测试在 Linux 上仍检查 0600,Windows 上只检查文件写出来了。
|
||||
- `tests/test_vedastro_runtime_ops.py` 要求已安装的 VedAstro 是 1.23.25,本机是 1.23.30。与这次缓存改动无关。
|
||||
- `tests/test_three_engine_parity_runner.py` 的公开盘回放把 PyJHora 记成 raw_imported,不是 structured_captured,同时在下载星历。运行器没改。
|
||||
|
||||
## 环境缺口
|
||||
|
||||
- 生产和测试环境里成功缓存的实际秒数,这份仓库读不到。部署后看状态接口的两个期限数字即可。
|
||||
- 生产卷上已经堆了多少文件,这里不知道。
|
||||
- 主测量挡住了 VedAstro 导入。本机有 SDK,官方端口能连通。不挡 SDK 的探针仍关闭外网取数,一次完整排盘 4.167 秒,没有报错,没有拉真实快照。
|
||||
|
||||
## 未做
|
||||
|
||||
- 没有改部署文件、生产环境变量、数据库,也没有手工清生产卷。
|
||||
- 没有把咨询卡片缓存并进周期清理。
|
||||
@@ -470,7 +470,7 @@
|
||||
| `TASK-rect-scoring-perf-20261013.md` | — | 校正打分提速:引导窗口平方扫描、候选不变量重复算、响应带无读取点字段、候选数上限 241(BUG-1323~1325、1345) | 待领取 | `codex/rect-scoring-perf-20261013` |
|
||||
| `TASK-node-runtime-resources-20261013.md` | — | 报告 worker abort 监听泄漏、本地 PG 适配层往返与元数据缓存、无上限内存表、用户端会话读缓存(BUG-1326~1328、1346,PG 与缓存部分 A/B) | 待领取 | `codex/node-runtime-resources-20261013` |
|
||||
| `TASK-home-sidebar-rerender-20261013.md` | — | 流式回答时侧栏每帧整栏重渲染(BUG-1329,A/B) | 待领取 | `codex/home-sidebar-rerender-20261013` |
|
||||
| `TASK-api-disk-cache-retention-20261013.md` | — | API 磁盘缓存只进不出、VedAstro 审计文件留存出生资料、失败快照久缓存、缓存键规范化(BUG-1330~1332,A/B) | 待领取 | `codex/api-disk-cache-retention-20261013` |
|
||||
| `TASK-api-disk-cache-retention-20261013.md` | `PROGRESS-api-disk-cache-retention-20261013.md` | API 磁盘缓存只进不出、VedAstro 审计文件留存出生资料、失败快照久缓存、缓存键规范化(BUG-1330~1332,A/B) | 待验收 | `codex/api-disk-cache-retention-20261013` |
|
||||
| `TASK-report-reference-perf-20261013.md` | — | 长报告出稿提速:术语替换命中预检、哈希单次遍历、同盘复用(BUG-1335~1336,A/B) | 待领取 | `codex/report-reference-perf-20261013` |
|
||||
| `TASK-consult-prep-parallel-20261013.md` | — | 普通对话发问前准备步骤并行、模型版本短缓存、多领域并发开关(BUG-1337~1339,A/B) | 待领取 | `codex/consult-prep-parallel-20261013` |
|
||||
| `TASK-frontend-dead-code-20261013.md` | — | 前端死代码 17 文件与无调用方接口删除,含未登录重计算接口(BUG-1342) | 待领取 | `codex/frontend-dead-code-20261013` |
|
||||
|
||||
@@ -0,0 +1,66 @@
|
||||
# 磁盘缓存留存 A/B(2026-10-14)
|
||||
|
||||
对照的是本分支的父提交 `66cd60081f37a57680274633fcb3fb942d01bb2f`(建分支时的 `origin/staging`)。任务书里写的 `77057a819` 已经不是当前测试线头。A 用 `git show` 把父提交的接口文件载入同一进程,并把它的仓库路径指回本工作树,否则临时副本找不到引擎。B 是本分支。
|
||||
|
||||
演练用虚构盘,不写真实出生资料。文件都在系统临时目录,不写工作树里的 `scratch/`。
|
||||
|
||||
主测量挡住了 `jhora` 和 `vedastro` 的导入。本机装了 VedAstro SDK,并且能连上官方端口;另有一组不挡 SDK、但仍关闭外网取数,见文末。
|
||||
|
||||
## 命令
|
||||
|
||||
```powershell
|
||||
.venv\Scripts\python.exe scripts\research\api_chart_cache_byte_diff.py
|
||||
.venv\Scripts\python.exe scripts\research\api_disk_cache_ab.py
|
||||
.venv\Scripts\python.exe scripts\research\api_disk_cache_ab.py --with-sdk --sdk-probe
|
||||
```
|
||||
|
||||
## 命中率
|
||||
|
||||
20 张虚构盘 × 3 种写法 × 快速/完整 × 7 天 × 每种重复 3 次 = 2520 次取键。取键用的是真实缓存键函数,没有把 2520 次都做成完整排盘。
|
||||
|
||||
| | 改前 A | 改后 B | 变化 |
|
||||
| --- | --- | --- | --- |
|
||||
| 不同缓存键 | 840 | 280 | 少 560 |
|
||||
| 命中次数 | 1680 | 2240 | 多 560 |
|
||||
| 命中率 | 0.6667 | 0.8889 | B / A = 1.333 |
|
||||
|
||||
目标是混合写法下 B 至少为 A 的 2.5 倍。按任务书规定的「重复 3 次」算,没有达到:A 自己靠重复就已经有三分之二命中,写法合并只能再抬到约 89%。若不算这 3 次重复、只看三种写法各问一次,A 的命中率是 0,B 是 0.6667。目标数字不改。
|
||||
|
||||
## 七天后的文件
|
||||
|
||||
留存演练写的是短记录,用来数文件,不是完整盘面。期限 900 秒,只有当天的键还没过期。
|
||||
|
||||
| | 改前 A | 改后 B |
|
||||
| --- | --- | --- |
|
||||
| 清理前文件数 | 840 | 280 |
|
||||
| 清理前字节 | 109200 | 36400 |
|
||||
| 清理后文件数 | 840(旧代码没有这份清理) | 40 |
|
||||
| 当天仍有效的键 | 120 | 40 |
|
||||
|
||||
B 清理后的文件数等于当天有效键数,达到「不超过当天活跃键」的目标。
|
||||
|
||||
审计文件:B 在默认关闭时写出 0 个。达到目标。
|
||||
|
||||
负缓存:种入当天和前一天各一份,清理后剩 1 份(当天)。达到目标。
|
||||
|
||||
## 清理耗时
|
||||
|
||||
| 场景 | 结果 |
|
||||
| --- | --- |
|
||||
| 预置 10000 个已过期排盘缓存后,整次清理 | 20.174 秒,排盘目录 20170.902 毫秒,删掉 10000,剩下 0 |
|
||||
| 上面七天演练里的排盘目录 | 370.161 毫秒 |
|
||||
|
||||
目标是 10000 个文件在 2 秒内、且不在请求线程。本次在后台函数里测,不在请求线程;耗时没有进 2 秒。这是 Windows 上逐个读文件再删除的实测,目标不改。默认清理间隔仍是 1800 秒,失败快照短期限仍是 300 秒,不因为这次没进 2 秒就改默认值。
|
||||
|
||||
## 排盘耗时(一次虚构盘,快速排盘,外网关闭)
|
||||
|
||||
| | 中位数 |
|
||||
| --- | --- |
|
||||
| 未命中 3 次 | 23.036 毫秒 |
|
||||
| 命中 3 次 | 5.249 毫秒 |
|
||||
|
||||
## 响应是否还是同一张盘
|
||||
|
||||
数值等价的写法(小时 12 / 12.0 / "12",纬度数字和字符串,时区 5.5 和 "5.5",秒缺省和 0)在快速排盘下去掉 `runtime_cache` 后逐字节相同。完整排盘只差官方快照缓存上的写入时刻和到期时刻,盘面没有别的差异。父提交和本分支对同一张快速排盘、去掉 `runtime_cache` 后逐字节相同。因此这些字段可以合并进同一把缓存键。
|
||||
|
||||
不挡 SDK 的一组:本机有 VedAstro SDK,官方端口能连通。探针仍关闭外网取数。一次完整排盘 4.167 秒,没有报错。没有拉取真实快照。
|
||||
@@ -0,0 +1,609 @@
|
||||
"""Retention rules for scratch/local caches.
|
||||
|
||||
Counts only. Never prints file names or cache payloads.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
UNREADABLE_KEEP_SECONDS = 24 * 60 * 60
|
||||
_HEX64_JSON = re.compile(r"^[0-9a-f]{64}\.json$")
|
||||
_ARTIFACT_JSON = re.compile(r"^[A-Za-z0-9_]+-[0-9a-f]{12}-[0-9a-f]{12}\.json$")
|
||||
_GATEWAY_JOB_JSON = re.compile(r"^vgw_[0-9a-f]{20}\.json$")
|
||||
_PRUNE_THREAD_NAME = "jyotish-scratch-prune"
|
||||
|
||||
# Directory names under scratch/local that this module already knows how to prune.
|
||||
# Subdirectories of vedastro_adapter are listed so a future writer is not a new top-level dir.
|
||||
REGISTERED_DIR_NAMES = frozenset(
|
||||
{
|
||||
"api_chart_cache",
|
||||
"high_rigor_jobs",
|
||||
"api_chart_response_jobs",
|
||||
"async_jobs.sqlite3",
|
||||
"vedastro_adapter",
|
||||
"vedastro_snapshot_cache",
|
||||
"vedastro_snapshot_negative_cache",
|
||||
"vedastro_gateway_jobs",
|
||||
"varga_segments",
|
||||
}
|
||||
)
|
||||
|
||||
# Offline tools or caches that already prune themselves. Not a new retention rule.
|
||||
OFFLINE_SCRATCH_NAMES = frozenset(
|
||||
{
|
||||
"consult_full_data_cache",
|
||||
"ocr_extract",
|
||||
"three_engine_parity",
|
||||
"public_release_deny_patterns.txt",
|
||||
"pdf_review_123456",
|
||||
}
|
||||
)
|
||||
|
||||
REGISTRY: tuple[dict[str, Any], ...] = (
|
||||
{
|
||||
"name": "api_chart_cache",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "created_at plus chart TTL, or unreadable json older than 24h",
|
||||
},
|
||||
{
|
||||
"name": "high_rigor_jobs",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "expires_at_unix has passed; bad json is kept",
|
||||
},
|
||||
{
|
||||
"name": "api_chart_response_jobs",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "expires_at_unix has passed; bad json is kept",
|
||||
},
|
||||
{
|
||||
"name": "async_jobs.sqlite3",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "sqlite rows whose expires_at has passed",
|
||||
},
|
||||
{
|
||||
"name": "vedastro_adapter",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "audit artifacts, response_cache, and official_full_snapshot_cache",
|
||||
},
|
||||
{
|
||||
"name": "vedastro_snapshot_cache",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "reference_date before today UTC minus MAX_STALE_DAYS minus 1",
|
||||
},
|
||||
{
|
||||
"name": "vedastro_snapshot_negative_cache",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "reference_date before today UTC",
|
||||
},
|
||||
{
|
||||
"name": "vedastro_gateway_jobs",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "completed or failed jobs older than the keep window, plus their raw archive",
|
||||
},
|
||||
{
|
||||
"name": "varga_segments",
|
||||
"holds_birth_data": True,
|
||||
"deletable": "mtime older than JYOTISH_VARGA_SEGMENTS_KEEP_DAYS",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def write_private_json(path: Path, payload: Any) -> None:
|
||||
"""Atomic UTF-8 JSON, mode 0600. Compact, stable key order."""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temp = path.with_name(f"{path.name}.{secrets.token_hex(8)}.tmp")
|
||||
temp.write_text(
|
||||
json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")),
|
||||
encoding="utf-8",
|
||||
)
|
||||
try:
|
||||
os.chmod(temp, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
os.replace(temp, path)
|
||||
try:
|
||||
os.chmod(path, 0o600)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _finite_float(value: Any) -> float | None:
|
||||
if isinstance(value, bool) or value is None:
|
||||
return None
|
||||
if isinstance(value, (int, float)):
|
||||
number = float(value)
|
||||
elif isinstance(value, str):
|
||||
text = value.strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
number = float(text)
|
||||
except ValueError:
|
||||
return None
|
||||
else:
|
||||
return None
|
||||
if not math.isfinite(number):
|
||||
return None
|
||||
return number
|
||||
|
||||
|
||||
def _exact_int(value: Any) -> int | None:
|
||||
if isinstance(value, bool) or value is None:
|
||||
return None
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
if isinstance(value, float):
|
||||
if not math.isfinite(value) or not value.is_integer():
|
||||
return None
|
||||
return int(value)
|
||||
if isinstance(value, str) and re.fullmatch(r"[+-]?\d+", value.strip()):
|
||||
return int(value.strip())
|
||||
return None
|
||||
|
||||
|
||||
def normalize_chart_cache_birth(body: dict | None) -> dict[str, Any]:
|
||||
"""Cache-key birth fields. Unparseable values stay as they arrived."""
|
||||
payload = body if isinstance(body, dict) else {}
|
||||
normalized: dict[str, Any] = {}
|
||||
for key in ("year", "month", "day"):
|
||||
parsed = _exact_int(payload.get(key))
|
||||
normalized[key] = payload.get(key) if parsed is None else parsed
|
||||
for key in ("hour", "minute", "lat", "lon", "tz"):
|
||||
raw = payload.get(key)
|
||||
parsed = _finite_float(raw)
|
||||
normalized[key] = raw if parsed is None else parsed
|
||||
second_raw = payload.get("second", 0)
|
||||
second = _finite_float(second_raw)
|
||||
normalized["second"] = second_raw if second is None else second
|
||||
return normalized
|
||||
|
||||
|
||||
def normalize_chart_cache_day(value: Any) -> Any:
|
||||
if value is None or value == "":
|
||||
return value
|
||||
return str(value)[:10]
|
||||
|
||||
|
||||
def _blank() -> dict[str, int]:
|
||||
return {"scanned": 0, "removed": 0, "errors": 0}
|
||||
|
||||
|
||||
def _older_than(path: Path, moment: float, seconds: float) -> bool:
|
||||
try:
|
||||
return moment - path.stat().st_mtime > seconds
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
def _unlink(path: Path, stats: dict[str, int]) -> None:
|
||||
try:
|
||||
path.unlink()
|
||||
stats["removed"] += 1
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
|
||||
|
||||
def _repo_root() -> Path:
|
||||
return ROOT
|
||||
|
||||
|
||||
def _api_module(api_module=None):
|
||||
if api_module is not None:
|
||||
return api_module
|
||||
import jyotish_api_server as api
|
||||
|
||||
return api
|
||||
|
||||
|
||||
def prune_async_job_records(now: float | None = None, api_module=None) -> dict[str, Any]:
|
||||
"""Same file and sqlite rules as the old startup cleanup. Bad job JSON is kept."""
|
||||
api = _api_module(api_module)
|
||||
moment = time.time() if now is None else float(now)
|
||||
scanned = 0
|
||||
removed = 0
|
||||
errors = 0
|
||||
directories: dict[str, dict[str, int]] = {}
|
||||
if api._async_job_backend() == "sqlite":
|
||||
try:
|
||||
with api._sqlite_job_connection() as connection:
|
||||
scanned = int(connection.execute("SELECT COUNT(*) FROM async_jobs").fetchone()[0])
|
||||
deleted = connection.execute(
|
||||
"DELETE FROM async_jobs WHERE expires_at IS NOT NULL AND expires_at <= ?",
|
||||
(moment,),
|
||||
).rowcount
|
||||
removed = int(deleted or 0)
|
||||
except OSError:
|
||||
errors += 1
|
||||
directories["high_rigor_jobs"] = _blank()
|
||||
directories["api_chart_response_jobs"] = _blank()
|
||||
directories["async_jobs_sqlite"] = {"scanned": scanned, "removed": removed, "errors": errors}
|
||||
return {
|
||||
"scope": "async_job_cleanup",
|
||||
"scanned": scanned,
|
||||
"removed": removed,
|
||||
"errors": errors,
|
||||
"directories": directories,
|
||||
}
|
||||
|
||||
for scope, label in (
|
||||
(api._HIGH_RIGOR_JOB_SCOPE, "high_rigor_jobs"),
|
||||
(api._API_CHART_CACHE_SCOPE, "api_chart_response_jobs"),
|
||||
):
|
||||
stats = _blank()
|
||||
try:
|
||||
directory = api._async_job_dir(scope)
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
directories[label] = stats
|
||||
errors += stats["errors"]
|
||||
continue
|
||||
if directory.is_dir():
|
||||
for path in directory.glob("*.json"):
|
||||
stats["scanned"] += 1
|
||||
try:
|
||||
record = json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
continue
|
||||
expires_at = record.get("expires_at_unix") if isinstance(record, dict) else None
|
||||
if isinstance(expires_at, (int, float)) and not isinstance(expires_at, bool) and moment >= float(expires_at):
|
||||
_unlink(path, stats)
|
||||
directories[label] = stats
|
||||
scanned += stats["scanned"]
|
||||
removed += stats["removed"]
|
||||
errors += stats["errors"]
|
||||
return {
|
||||
"scope": "async_job_cleanup",
|
||||
"scanned": scanned,
|
||||
"removed": removed,
|
||||
"errors": errors,
|
||||
"directories": directories,
|
||||
}
|
||||
|
||||
|
||||
def _prune_chart_cache(moment: float, api_module=None) -> dict[str, int]:
|
||||
api = _api_module(api_module)
|
||||
stats = _blank()
|
||||
try:
|
||||
directory = api._api_chart_cache_dir()
|
||||
ttl = api._api_chart_cache_ttl_seconds()
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
return stats
|
||||
if not directory.is_dir():
|
||||
return stats
|
||||
for path in directory.glob("*.json"):
|
||||
if not _HEX64_JSON.match(path.name):
|
||||
continue
|
||||
stats["scanned"] += 1
|
||||
try:
|
||||
record = json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
if _older_than(path, moment, UNREADABLE_KEEP_SECONDS):
|
||||
_unlink(path, stats)
|
||||
continue
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
continue
|
||||
created_at = record.get("created_at") if isinstance(record, dict) else None
|
||||
if not isinstance(created_at, (int, float)) or isinstance(created_at, bool):
|
||||
if _older_than(path, moment, UNREADABLE_KEEP_SECONDS):
|
||||
_unlink(path, stats)
|
||||
continue
|
||||
if moment - float(created_at) > ttl:
|
||||
_unlink(path, stats)
|
||||
return stats
|
||||
|
||||
|
||||
def _adapter():
|
||||
import vedastro_service_adapter as adapter
|
||||
|
||||
return adapter
|
||||
|
||||
|
||||
def _prune_created_at_dir(directory: Path, moment: float, ttl_for_record: Callable[[dict], float]) -> dict[str, int]:
|
||||
stats = _blank()
|
||||
if not directory.is_dir():
|
||||
return stats
|
||||
for path in directory.glob("*.json"):
|
||||
if not _HEX64_JSON.match(path.name):
|
||||
continue
|
||||
stats["scanned"] += 1
|
||||
try:
|
||||
record = json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
if _older_than(path, moment, UNREADABLE_KEEP_SECONDS):
|
||||
_unlink(path, stats)
|
||||
continue
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
continue
|
||||
created_at = record.get("created_at") if isinstance(record, dict) else None
|
||||
if not isinstance(created_at, (int, float)) or isinstance(created_at, bool):
|
||||
if _older_than(path, moment, UNREADABLE_KEEP_SECONDS):
|
||||
_unlink(path, stats)
|
||||
continue
|
||||
try:
|
||||
ttl = float(ttl_for_record(record))
|
||||
except (TypeError, ValueError):
|
||||
stats["errors"] += 1
|
||||
continue
|
||||
if moment - float(created_at) > ttl:
|
||||
_unlink(path, stats)
|
||||
return stats
|
||||
|
||||
|
||||
def _prune_response_cache(moment: float) -> dict[str, int]:
|
||||
adapter = _adapter()
|
||||
return _prune_created_at_dir(
|
||||
Path(adapter.ARTIFACT_DIR) / "response_cache",
|
||||
moment,
|
||||
lambda _record: adapter._cache_ttl_seconds(),
|
||||
)
|
||||
|
||||
|
||||
def _prune_snapshot_semantic_cache(moment: float) -> dict[str, int]:
|
||||
adapter = _adapter()
|
||||
|
||||
def ttl_for(record: dict) -> float:
|
||||
if record.get("negative") is True:
|
||||
return adapter._official_full_snapshot_negative_ttl_seconds()
|
||||
return adapter._official_full_snapshot_cache_ttl_seconds()
|
||||
|
||||
return _prune_created_at_dir(
|
||||
Path(adapter.ARTIFACT_DIR) / "official_full_snapshot_cache",
|
||||
moment,
|
||||
ttl_for,
|
||||
)
|
||||
|
||||
|
||||
def _audit_keep_seconds() -> float:
|
||||
raw = os.environ.get("VEDASTRO_AUDIT_ARTIFACT_KEEP_DAYS", "7").strip()
|
||||
try:
|
||||
days = float(raw) if raw else 7.0
|
||||
except ValueError:
|
||||
days = 7.0
|
||||
return max(0.0, days) * 86400.0
|
||||
|
||||
|
||||
def _prune_audit_artifacts(moment: float) -> dict[str, int]:
|
||||
adapter = _adapter()
|
||||
stats = _blank()
|
||||
directory = Path(adapter.ARTIFACT_DIR)
|
||||
if not directory.is_dir():
|
||||
return stats
|
||||
enabled = adapter._audit_artifacts_enabled()
|
||||
max_age = _audit_keep_seconds() if enabled else UNREADABLE_KEEP_SECONDS
|
||||
for path in directory.glob("*.json"):
|
||||
if not path.is_file() or not _ARTIFACT_JSON.match(path.name):
|
||||
continue
|
||||
stats["scanned"] += 1
|
||||
if _older_than(path, moment, max_age):
|
||||
_unlink(path, stats)
|
||||
return stats
|
||||
|
||||
|
||||
def _parse_day(value: Any):
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
raw = value[:10]
|
||||
try:
|
||||
return datetime.strptime(raw, "%Y-%m-%d").date()
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _prune_reference_dated(directory: Path, moment: float, *, earliest_kept) -> dict[str, int]:
|
||||
stats = _blank()
|
||||
if not directory.is_dir():
|
||||
return stats
|
||||
for path in directory.glob("*.json"):
|
||||
if not _HEX64_JSON.match(path.name):
|
||||
continue
|
||||
stats["scanned"] += 1
|
||||
try:
|
||||
record = json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
if _older_than(path, moment, UNREADABLE_KEEP_SECONDS):
|
||||
_unlink(path, stats)
|
||||
continue
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
continue
|
||||
ref = _parse_day(record.get("reference_date") if isinstance(record, dict) else None)
|
||||
if ref is None:
|
||||
if _older_than(path, moment, UNREADABLE_KEEP_SECONDS):
|
||||
_unlink(path, stats)
|
||||
continue
|
||||
if ref < earliest_kept:
|
||||
_unlink(path, stats)
|
||||
return stats
|
||||
|
||||
|
||||
def _prune_vedastro_snapshot_cache(moment: float) -> dict[str, int]:
|
||||
from vedastro_snapshot_cache import MAX_STALE_DAYS
|
||||
|
||||
today = datetime.fromtimestamp(moment, timezone.utc).date()
|
||||
earliest = today - timedelta(days=int(MAX_STALE_DAYS) + 1)
|
||||
raw = os.environ.get("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", "").strip()
|
||||
directory = Path(raw) if raw else _repo_root() / "scratch" / "local" / "vedastro_snapshot_cache"
|
||||
return _prune_reference_dated(directory, moment, earliest_kept=earliest)
|
||||
|
||||
|
||||
def _prune_vedastro_negative_cache(moment: float) -> dict[str, int]:
|
||||
today = datetime.fromtimestamp(moment, timezone.utc).date()
|
||||
raw = os.environ.get("JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR", "").strip()
|
||||
directory = Path(raw) if raw else _repo_root() / "scratch" / "local" / "vedastro_snapshot_negative_cache"
|
||||
return _prune_reference_dated(directory, moment, earliest_kept=today)
|
||||
|
||||
|
||||
def _parse_updated_at(value: Any) -> float | None:
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
return None
|
||||
text = value.strip().replace("Z", "+00:00")
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text)
|
||||
except ValueError:
|
||||
return None
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||||
return parsed.timestamp()
|
||||
|
||||
|
||||
def _gateway_keep_seconds() -> float:
|
||||
raw = os.environ.get("VEDASTRO_GATEWAY_JOB_KEEP_SECONDS", "86400").strip()
|
||||
try:
|
||||
return max(0.0, float(raw) if raw else 86400.0)
|
||||
except ValueError:
|
||||
return 86400.0
|
||||
|
||||
|
||||
def _prune_gateway_jobs(moment: float) -> dict[str, int]:
|
||||
stats = _blank()
|
||||
raw = os.environ.get("VEDASTRO_GATEWAY_QUEUE_DIR", "").strip()
|
||||
directory = Path(raw).expanduser() if raw else _repo_root() / "scratch" / "local" / "vedastro_gateway_jobs"
|
||||
if not directory.is_dir():
|
||||
return stats
|
||||
keep = _gateway_keep_seconds()
|
||||
for path in directory.glob("*.json"):
|
||||
if not _GATEWAY_JOB_JSON.match(path.name):
|
||||
continue
|
||||
stats["scanned"] += 1
|
||||
try:
|
||||
record = json.loads(path.read_text(encoding="utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
except OSError:
|
||||
stats["errors"] += 1
|
||||
continue
|
||||
if not isinstance(record, dict):
|
||||
continue
|
||||
status = str(record.get("status") or "")
|
||||
if status not in {"completed", "failed"}:
|
||||
continue
|
||||
updated = _parse_updated_at(record.get("updated_at"))
|
||||
if updated is None or moment - updated <= keep:
|
||||
continue
|
||||
companion = directory / f"{path.stem}.official_raw_response.json"
|
||||
_unlink(path, stats)
|
||||
if companion.is_file():
|
||||
stats["scanned"] += 1
|
||||
_unlink(companion, stats)
|
||||
return stats
|
||||
|
||||
|
||||
def _varga_keep_seconds() -> float:
|
||||
raw = os.environ.get("JYOTISH_VARGA_SEGMENTS_KEEP_DAYS", "7").strip()
|
||||
try:
|
||||
days = float(raw) if raw else 7.0
|
||||
except ValueError:
|
||||
days = 7.0
|
||||
return max(0.0, days) * 86400.0
|
||||
|
||||
|
||||
def _prune_varga_segments(moment: float) -> dict[str, int]:
|
||||
stats = _blank()
|
||||
raw = os.environ.get("JYOTISH_CACHE_DIR", "").strip()
|
||||
root = Path(raw) if raw else _repo_root() / "scratch" / "local"
|
||||
directory = root / "varga_segments"
|
||||
if not directory.is_dir():
|
||||
return stats
|
||||
max_age = _varga_keep_seconds()
|
||||
for path in directory.glob("*.json"):
|
||||
if not _HEX64_JSON.match(path.name):
|
||||
continue
|
||||
stats["scanned"] += 1
|
||||
if _older_than(path, moment, max_age):
|
||||
_unlink(path, stats)
|
||||
return stats
|
||||
|
||||
|
||||
def _stamp(stats: dict[str, int], started: float) -> dict[str, Any]:
|
||||
return {
|
||||
"scanned": int(stats.get("scanned", 0) or 0),
|
||||
"removed": int(stats.get("removed", 0) or 0),
|
||||
"errors": int(stats.get("errors", 0) or 0),
|
||||
"elapsed_ms": round((time.perf_counter() - started) * 1000, 3),
|
||||
}
|
||||
|
||||
|
||||
def prune_all(now: float | None = None, api_module=None) -> dict[str, Any]:
|
||||
"""Prune every registered directory. Exceptions become error counts."""
|
||||
moment = time.time() if now is None else float(now)
|
||||
api = _api_module(api_module)
|
||||
started = time.perf_counter()
|
||||
directories: dict[str, dict[str, Any]] = {}
|
||||
|
||||
def _run(name: str, func: Callable[[float], dict[str, int]]) -> None:
|
||||
t0 = time.perf_counter()
|
||||
try:
|
||||
stats = func(moment)
|
||||
except Exception:
|
||||
stats = {"scanned": 0, "removed": 0, "errors": 1}
|
||||
directories[name] = _stamp(stats, t0)
|
||||
|
||||
_run("api_chart_cache", lambda moment: _prune_chart_cache(moment, api))
|
||||
job_started = time.perf_counter()
|
||||
try:
|
||||
job = prune_async_job_records(moment, api)
|
||||
job_dirs = job.get("directories") or {}
|
||||
except Exception:
|
||||
job_dirs = {
|
||||
"high_rigor_jobs": {"scanned": 0, "removed": 0, "errors": 1},
|
||||
"api_chart_response_jobs": _blank(),
|
||||
}
|
||||
job_elapsed = round((time.perf_counter() - job_started) * 1000, 3)
|
||||
for name, stats in job_dirs.items():
|
||||
stamped = _stamp(stats, job_started)
|
||||
stamped["elapsed_ms"] = job_elapsed
|
||||
directories[name] = stamped
|
||||
_run("vedastro_response_cache", _prune_response_cache)
|
||||
_run("vedastro_official_full_snapshot_cache", _prune_snapshot_semantic_cache)
|
||||
_run("vedastro_audit_artifacts", _prune_audit_artifacts)
|
||||
_run("vedastro_snapshot_cache", _prune_vedastro_snapshot_cache)
|
||||
_run("vedastro_snapshot_negative_cache", _prune_vedastro_negative_cache)
|
||||
_run("vedastro_gateway_jobs", _prune_gateway_jobs)
|
||||
_run("varga_segments", _prune_varga_segments)
|
||||
return {
|
||||
"scanned": sum(item["scanned"] for item in directories.values()),
|
||||
"removed": sum(item["removed"] for item in directories.values()),
|
||||
"errors": sum(item["errors"] for item in directories.values()),
|
||||
"elapsed_ms": round((time.perf_counter() - started) * 1000, 3),
|
||||
"directories": directories,
|
||||
}
|
||||
|
||||
|
||||
def start_periodic_prune(interval_s: float | None = None):
|
||||
"""Daemon thread. interval <= 0 starts nothing. Startup prune is separate."""
|
||||
if interval_s is None:
|
||||
raw = os.environ.get("JYOTISH_SCRATCH_PRUNE_INTERVAL_SECONDS", "1800").strip()
|
||||
try:
|
||||
interval_s = float(raw) if raw else 1800.0
|
||||
except ValueError:
|
||||
interval_s = 1800.0
|
||||
if interval_s <= 0:
|
||||
return None
|
||||
|
||||
def _loop() -> None:
|
||||
while True:
|
||||
time.sleep(interval_s)
|
||||
try:
|
||||
prune_all()
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
thread = threading.Thread(target=_loop, name=_PRUNE_THREAD_NAME, daemon=True)
|
||||
thread.start()
|
||||
return thread
|
||||
@@ -2590,24 +2590,17 @@ def _vedastro_runtime_fingerprint() -> dict:
|
||||
|
||||
|
||||
def _build_api_chart_cache_payload(body: dict) -> dict:
|
||||
import api_scratch_retention as retention
|
||||
|
||||
today = body.get('today') or body.get('current_date')
|
||||
return {
|
||||
'cache_schema_version': 5,
|
||||
'birth': {
|
||||
'year': body.get('year'),
|
||||
'month': body.get('month'),
|
||||
'day': body.get('day'),
|
||||
'hour': body.get('hour'),
|
||||
'minute': body.get('minute'),
|
||||
'second': body.get('second', 0),
|
||||
'lat': body.get('lat'),
|
||||
'lon': body.get('lon'),
|
||||
'tz': body.get('tz'),
|
||||
},
|
||||
'cache_schema_version': 6,
|
||||
'birth': retention.normalize_chart_cache_birth(body),
|
||||
'calculation': {
|
||||
'ayanamsa': _request_ayanamsa(body),
|
||||
'node_mode': body.get('node_mode', body.get('nodeMode', 'mean')),
|
||||
'today': body.get('today') or body.get('current_date'),
|
||||
'transit_date': body.get('transit_date'),
|
||||
'today': retention.normalize_chart_cache_day(today),
|
||||
'transit_date': retention.normalize_chart_cache_day(body.get('transit_date')),
|
||||
'skip_vedastro_main_entry_overview': bool(body.get('skip_vedastro_main_entry_overview')),
|
||||
},
|
||||
'vedastro_runtime': _vedastro_runtime_fingerprint(),
|
||||
@@ -2678,14 +2671,15 @@ def _store_api_chart_response_cache(cache_payload: dict, payload: dict) -> dict:
|
||||
)
|
||||
ttl_seconds = _api_chart_cache_ttl_seconds()
|
||||
if ttl_seconds > 0:
|
||||
record = {
|
||||
'cache_key': cache_key,
|
||||
'created_at': created_at_unix,
|
||||
'payload': payload_with_cache,
|
||||
}
|
||||
_api_chart_cache_path(cache_key).write_text(
|
||||
json.dumps(record, ensure_ascii=False, sort_keys=True),
|
||||
encoding='utf-8',
|
||||
import api_scratch_retention as retention
|
||||
|
||||
retention.write_private_json(
|
||||
_api_chart_cache_path(cache_key),
|
||||
{
|
||||
'cache_key': cache_key,
|
||||
'created_at': created_at_unix,
|
||||
'payload': payload_with_cache,
|
||||
},
|
||||
)
|
||||
return payload_with_cache
|
||||
|
||||
@@ -2738,30 +2732,9 @@ def _sqlite_job_connection() -> sqlite3.Connection:
|
||||
|
||||
def prune_expired_async_jobs() -> dict:
|
||||
"""Best-effort startup cleanup for local job records; never reads payloads."""
|
||||
removed = 0
|
||||
scanned = 0
|
||||
if _async_job_backend() == "sqlite":
|
||||
with _sqlite_job_connection() as connection:
|
||||
scanned = connection.execute("SELECT COUNT(*) FROM async_jobs").fetchone()[0]
|
||||
removed = connection.execute(
|
||||
"DELETE FROM async_jobs WHERE expires_at IS NOT NULL AND expires_at <= ?", (time.time(),)
|
||||
).rowcount
|
||||
return {'scope': 'async_job_cleanup', 'scanned': scanned, 'removed': removed}
|
||||
for scope in (_HIGH_RIGOR_JOB_SCOPE, _API_CHART_CACHE_SCOPE):
|
||||
directory = _async_job_dir(scope)
|
||||
if not directory.is_dir():
|
||||
continue
|
||||
for path in directory.glob('*.json'):
|
||||
scanned += 1
|
||||
try:
|
||||
record = json.loads(path.read_text(encoding='utf-8'))
|
||||
expires_at = record.get('expires_at_unix') if isinstance(record, dict) else None
|
||||
if isinstance(expires_at, (int, float)) and time.time() >= float(expires_at):
|
||||
path.unlink()
|
||||
removed += 1
|
||||
except (OSError, json.JSONDecodeError):
|
||||
continue
|
||||
return {'scope': 'async_job_cleanup', 'scanned': scanned, 'removed': removed}
|
||||
import api_scratch_retention as retention
|
||||
|
||||
return retention.prune_async_job_records(api_module=sys.modules[__name__])
|
||||
|
||||
|
||||
def _new_async_job_identity(prefix: str) -> dict:
|
||||
@@ -10902,7 +10875,10 @@ def _parse_allowed_hosts(value):
|
||||
|
||||
|
||||
def start_server(port=5200, host='127.0.0.1', allowed_origins=None, allowed_hosts=None):
|
||||
cleanup = prune_expired_async_jobs()
|
||||
import api_scratch_retention as retention
|
||||
|
||||
cleanup = retention.prune_all(api_module=sys.modules[__name__])
|
||||
retention.start_periodic_prune()
|
||||
server = ThreadingHTTPServer((host, port), JyotishAPIHandler)
|
||||
server.daemon_threads = True
|
||||
server.allowed_origins = allowed_origins or DEFAULT_ALLOWED_ORIGINS
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
"""Byte-compare numeric chart-cache spellings. See api_disk_cache_ab.py."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
|
||||
import api_disk_cache_ab as ab
|
||||
|
||||
|
||||
def main() -> dict:
|
||||
parent_error = None
|
||||
try:
|
||||
parent = ab._load_parent()
|
||||
except Exception as exc:
|
||||
parent = None
|
||||
parent_error = exc.__class__.__name__
|
||||
report = ab.run_byte_diff(parent or ab.current)
|
||||
report["parent_import"] = parent_error or "ok"
|
||||
report["parent"] = ab.PARENT
|
||||
return report
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
json.dump(main(), sys.stdout, ensure_ascii=False, indent=2)
|
||||
sys.stdout.write("\n")
|
||||
@@ -0,0 +1,442 @@
|
||||
"""A/B for chart-cache keys, retention, and response bytes.
|
||||
|
||||
Parent functions come from git show of the branch base (origin/staging at
|
||||
branch creation), not from the task text's older baseline. Scratch files
|
||||
stay in a temporary directory.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import socket
|
||||
import statistics
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
PARENT = "66cd60081f37a57680274633fcb3fb942d01bb2f"
|
||||
os.environ.setdefault("JYOTISH_SKIP_LOCAL_ENV", "1")
|
||||
os.environ.setdefault("VEDASTRO_ENABLE_NETWORK", "0")
|
||||
|
||||
_SDK_PRESENT = importlib.util.find_spec("vedastro") is not None
|
||||
|
||||
|
||||
def _network_reachable() -> bool:
|
||||
try:
|
||||
sock = socket.create_connection(("api.vedastro.org", 443), 3)
|
||||
except OSError:
|
||||
return False
|
||||
sock.close()
|
||||
return True
|
||||
|
||||
|
||||
_NETWORK = _network_reachable()
|
||||
|
||||
|
||||
class _Block:
|
||||
def find_spec(self, name, path=None, target=None):
|
||||
if name.split(".")[0] in {"jhora", "vedastro"}:
|
||||
raise ModuleNotFoundError(f"No module named '{name}'")
|
||||
return None
|
||||
|
||||
|
||||
if "--with-sdk" not in sys.argv:
|
||||
sys.meta_path.insert(0, _Block())
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
import api_scratch_retention as retention # noqa: E402
|
||||
import jyotish_api_server as current # noqa: E402
|
||||
import vedastro_service_adapter as adapter # noqa: E402
|
||||
|
||||
|
||||
def _load_parent():
|
||||
raw = subprocess.check_output(["git", "show", f"{PARENT}:scripts/jyotish_api_server.py"], text=True)
|
||||
raw = raw.replace(
|
||||
"SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))",
|
||||
f"SCRIPTS_DIR = {str(ROOT / 'scripts')!r}",
|
||||
)
|
||||
raw = raw.replace(
|
||||
"REPO_ROOT = os.path.abspath(os.path.join(SCRIPTS_DIR, '..'))",
|
||||
f"REPO_ROOT = {str(ROOT)!r}",
|
||||
)
|
||||
directory = Path(tempfile.mkdtemp(prefix="jyotish-cache-a-"))
|
||||
path = directory / "jyotish_api_server_parent.py"
|
||||
path.write_text(raw, encoding="utf-8")
|
||||
spec = importlib.util.spec_from_file_location("jyotish_api_server_parent", path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[spec.name] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def _handler(module):
|
||||
cls = module.JyotishAPIHandler
|
||||
return cls.__new__(cls)
|
||||
|
||||
|
||||
def _strip(payload: dict) -> str:
|
||||
cloned = json.loads(json.dumps(payload, default=str))
|
||||
cloned.pop("runtime_cache", None)
|
||||
return json.dumps(cloned, ensure_ascii=False, sort_keys=True, default=str)
|
||||
|
||||
|
||||
def _diff_paths(left: str, right: str, limit: int = 12) -> list[str]:
|
||||
a = json.loads(left)
|
||||
b = json.loads(right)
|
||||
|
||||
found: list[str] = []
|
||||
|
||||
def walk(one, two, prefix: str) -> None:
|
||||
if len(found) >= limit:
|
||||
return
|
||||
if type(one) is not type(two):
|
||||
found.append(prefix or "$")
|
||||
return
|
||||
if isinstance(one, dict):
|
||||
for key in sorted(set(one) | set(two)):
|
||||
if key not in one or key not in two:
|
||||
found.append(f"{prefix}.{key}")
|
||||
continue
|
||||
walk(one[key], two[key], f"{prefix}.{key}")
|
||||
elif isinstance(one, list):
|
||||
if len(one) != len(two):
|
||||
found.append(f"{prefix}[]")
|
||||
return
|
||||
for index, (item, other) in enumerate(zip(one, two)):
|
||||
walk(item, other, f"{prefix}[{index}]")
|
||||
elif one != two:
|
||||
found.append(prefix or "$")
|
||||
|
||||
walk(a, b, "")
|
||||
return found
|
||||
|
||||
|
||||
def _body(hour, lat, **extra) -> dict:
|
||||
payload = {
|
||||
"year": 1990,
|
||||
"month": 1,
|
||||
"day": 1,
|
||||
"minute": 0,
|
||||
"lon": 20.0,
|
||||
"tz": 5.5,
|
||||
"hour": hour,
|
||||
"lat": lat,
|
||||
"today": "2026-10-14",
|
||||
}
|
||||
payload.update(extra)
|
||||
return payload
|
||||
|
||||
|
||||
def run_byte_diff(parent=None) -> dict:
|
||||
os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "0"
|
||||
os.environ["VEDASTRO_WRITE_AUDIT_ARTIFACTS"] = "0"
|
||||
parent = parent or current
|
||||
groups = {
|
||||
"hour": [_body(12, 10.0), _body(12.0, 10.0), _body("12", 10.0)],
|
||||
"lat": [_body(12, 10.0), _body(12, "10.0")],
|
||||
"tz": [_body(12, 10.0), _body(12, 10.0, tz="5.5")],
|
||||
"second": [_body(12, 10.0), _body(12, 10.0, second=0)],
|
||||
}
|
||||
report = {"groups": {}, "stable_repeat": None}
|
||||
handler = _handler(current)
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
sample = handler._compute_chart_sync(_body(12, 10.0, skip_vedastro_main_entry_overview=True))
|
||||
except Exception as exc:
|
||||
return {"ok": False, "error": exc.__class__.__name__, "elapsed_s": round(time.perf_counter() - started, 3)}
|
||||
sample_s = round(time.perf_counter() - started, 3)
|
||||
repeat = handler._compute_chart_sync(_body(12, 10.0, skip_vedastro_main_entry_overview=True))
|
||||
report["stable_repeat"] = _strip(sample) == _strip(repeat)
|
||||
report["sample_seconds"] = sample_s
|
||||
if not report["stable_repeat"]:
|
||||
report["volatile_paths"] = _diff_paths(_strip(sample), _strip(repeat))
|
||||
for name, bodies in groups.items():
|
||||
for skip in (True, False):
|
||||
rendered = []
|
||||
for body in bodies:
|
||||
item = dict(body)
|
||||
if skip:
|
||||
item["skip_vedastro_main_entry_overview"] = True
|
||||
rendered.append(_strip(handler._compute_chart_sync(item)))
|
||||
same = len(set(rendered)) == 1
|
||||
entry = {"skip": skip, "same": same}
|
||||
if not same:
|
||||
entry["paths"] = _diff_paths(rendered[0], rendered[1])
|
||||
clocks = ("cache_created_at", "cache_expires_at", "called_at")
|
||||
entry["only_clock_fields"] = bool(entry["paths"]) and all(
|
||||
path.endswith(clocks) for path in entry["paths"]
|
||||
)
|
||||
else:
|
||||
entry["only_clock_fields"] = False
|
||||
report["groups"][f"{name}:skip={skip}"] = entry
|
||||
if parent is not current:
|
||||
a_handler = _handler(parent)
|
||||
b_handler = handler
|
||||
body = _body(12, 10.0, skip_vedastro_main_entry_overview=True)
|
||||
a_text = _strip(a_handler._compute_chart_sync(body))
|
||||
b_text = _strip(b_handler._compute_chart_sync(body))
|
||||
report["a_vs_b_same"] = a_text == b_text
|
||||
if a_text != b_text:
|
||||
report["a_vs_b_paths"] = _diff_paths(a_text, b_text)
|
||||
report["ok"] = all(item["same"] or item.get("only_clock_fields") for item in report["groups"].values())
|
||||
return report
|
||||
|
||||
|
||||
def _requests():
|
||||
charts = []
|
||||
for index in range(20):
|
||||
charts.append(
|
||||
{
|
||||
"year": 1990,
|
||||
"month": 1,
|
||||
"day": 1 + (index % 27),
|
||||
"minute": index % 60,
|
||||
"lon": float(index),
|
||||
"tz": 5.5,
|
||||
}
|
||||
)
|
||||
spellings = (
|
||||
{"hour": 12, "lat": 10.0},
|
||||
{"hour": 12.0, "lat": "10.0"},
|
||||
{"hour": "12", "lat": 10.0},
|
||||
)
|
||||
days = [f"2026-10-{day:02d}" for day in range(8, 15)]
|
||||
rows = []
|
||||
for chart in charts:
|
||||
for spelling in spellings:
|
||||
for skip in (False, True):
|
||||
for day in days:
|
||||
body = dict(chart)
|
||||
body.update(spelling)
|
||||
body["today"] = day
|
||||
if skip:
|
||||
body["skip_vedastro_main_entry_overview"] = True
|
||||
rows.append(body)
|
||||
return rows
|
||||
|
||||
|
||||
def _key(module, body) -> str:
|
||||
return module._api_chart_cache_key(module._build_api_chart_cache_payload(body))
|
||||
|
||||
|
||||
def _hit_rate(module, rows) -> dict:
|
||||
seen = set()
|
||||
hits = 0
|
||||
for body in rows:
|
||||
for _repeat in range(3):
|
||||
key = _key(module, body)
|
||||
if key in seen:
|
||||
hits += 1
|
||||
else:
|
||||
seen.add(key)
|
||||
total = len(rows) * 3
|
||||
return {"requests": total, "unique_keys": len(seen), "hits": hits, "hit_rate": round(hits / total, 4)}
|
||||
|
||||
|
||||
def _seed_and_prune(module, rows) -> dict:
|
||||
now = time.time()
|
||||
os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "900"
|
||||
scratch = Path(tempfile.mkdtemp(prefix="jyotish-cache-b-"))
|
||||
chart = scratch / "api_chart_cache"
|
||||
jobs = scratch / "jobs"
|
||||
adapter_dir = scratch / "adapter"
|
||||
chart.mkdir()
|
||||
jobs.mkdir()
|
||||
current._api_chart_cache_dir = lambda: chart
|
||||
current._async_job_dir = lambda scope: jobs
|
||||
current._async_job_backend = lambda: "file"
|
||||
adapter.ARTIFACT_DIR = adapter_dir
|
||||
os.environ["JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR"] = str(scratch / "snap")
|
||||
os.environ["JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR"] = str(scratch / "neg")
|
||||
os.environ["VEDASTRO_GATEWAY_QUEUE_DIR"] = str(scratch / "gw")
|
||||
os.environ["JYOTISH_CACHE_DIR"] = str(scratch / "cache")
|
||||
days = sorted({body["today"] for body in rows})
|
||||
newest = days[-1]
|
||||
written = {}
|
||||
for body in rows:
|
||||
key = _key(module, body)
|
||||
age_days = days.index(body["today"])
|
||||
age_days = len(days) - 1 - age_days
|
||||
written[key] = now - age_days * 86400
|
||||
for key, created in written.items():
|
||||
path = chart / f"{key}.json"
|
||||
path.write_text(json.dumps({"cache_key": key, "created_at": created, "payload": {}}), encoding="utf-8")
|
||||
before = len(list(chart.glob("*.json")))
|
||||
before_bytes = sum(path.stat().st_size for path in chart.glob("*.json"))
|
||||
today_keys = len({_key(module, body) for body in rows if body["today"] == newest})
|
||||
if module is current:
|
||||
prune = retention.prune_all(now=now)
|
||||
after = len(list(chart.glob("*.json")))
|
||||
elapsed = prune["directories"]["api_chart_cache"]["elapsed_ms"]
|
||||
else:
|
||||
after = before
|
||||
elapsed = None
|
||||
return {
|
||||
"files_before": before,
|
||||
"bytes_before": before_bytes,
|
||||
"files_after": after,
|
||||
"today_active_keys": today_keys,
|
||||
"chart_prune_ms": elapsed,
|
||||
}
|
||||
|
||||
|
||||
def _artifact_counts() -> dict:
|
||||
scratch = Path(tempfile.mkdtemp(prefix="jyotish-artifact-ab-"))
|
||||
adapter.ARTIFACT_DIR = scratch
|
||||
os.environ.pop("VEDASTRO_WRITE_AUDIT_ARTIFACTS", None)
|
||||
result = {
|
||||
"operation": "official_full_snapshot",
|
||||
"source_metadata": {
|
||||
"operation": "official_full_snapshot",
|
||||
"request_hash": "a" * 64,
|
||||
"response_hash": "b" * 64,
|
||||
},
|
||||
}
|
||||
adapter._write_artifact(result)
|
||||
return {"b_artifact_files": len(list(scratch.glob("*.json")))}
|
||||
|
||||
|
||||
def _negative_cache_after_prune() -> dict:
|
||||
scratch = Path(tempfile.mkdtemp(prefix="jyotish-neg-ab-"))
|
||||
os.environ["JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR"] = str(scratch)
|
||||
os.environ["JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR"] = str(scratch / "unused-snap")
|
||||
os.environ["VEDASTRO_GATEWAY_QUEUE_DIR"] = str(scratch / "unused-gw")
|
||||
os.environ["JYOTISH_CACHE_DIR"] = str(scratch / "unused-cache")
|
||||
current._api_chart_cache_dir = lambda: scratch / "unused-chart"
|
||||
current._async_job_dir = lambda scope: scratch / "unused-jobs"
|
||||
(scratch / "unused-chart").mkdir()
|
||||
(scratch / "unused-jobs").mkdir()
|
||||
today = time.strftime("%Y-%m-%d", time.gmtime())
|
||||
yesterday = time.strftime("%Y-%m-%d", time.gmtime(time.time() - 86400))
|
||||
for label, day in (("a" * 64, today), ("b" * 64, yesterday)):
|
||||
path = scratch / f"{label}.json"
|
||||
path.write_text(json.dumps({"reference_date": day}), encoding="utf-8")
|
||||
retention.prune_all(now=time.time())
|
||||
return {"files_after": len(list(scratch.glob("*.json")))}
|
||||
|
||||
|
||||
def _prune_10k() -> dict:
|
||||
scratch = Path(tempfile.mkdtemp(prefix="jyotish-prune-10k-"))
|
||||
chart = scratch / "chart"
|
||||
chart.mkdir()
|
||||
current._api_chart_cache_dir = lambda: chart
|
||||
current._async_job_dir = lambda scope: scratch / "jobs"
|
||||
(scratch / "jobs").mkdir()
|
||||
current._async_job_backend = lambda: "file"
|
||||
adapter.ARTIFACT_DIR = scratch / "adapter"
|
||||
os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "900"
|
||||
os.environ["JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR"] = str(scratch / "snap")
|
||||
os.environ["JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR"] = str(scratch / "neg")
|
||||
os.environ["VEDASTRO_GATEWAY_QUEUE_DIR"] = str(scratch / "gw")
|
||||
os.environ["JYOTISH_CACHE_DIR"] = str(scratch / "cache")
|
||||
for index in range(10000):
|
||||
path = chart / f"{index:064x}.json"
|
||||
path.write_text('{"created_at":1,"payload":{}}', encoding="utf-8")
|
||||
started = time.perf_counter()
|
||||
result = retention.prune_all(now=time.time())
|
||||
elapsed = time.perf_counter() - started
|
||||
return {
|
||||
"prune_all_seconds": round(elapsed, 3),
|
||||
"chart_removed": result["directories"]["api_chart_cache"]["removed"],
|
||||
"chart_elapsed_ms": result["directories"]["api_chart_cache"]["elapsed_ms"],
|
||||
"files_left": len(list(chart.glob("*.json"))),
|
||||
}
|
||||
|
||||
|
||||
def _one_chart_timing() -> dict:
|
||||
os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "0"
|
||||
handler = _handler(current)
|
||||
body = _body(12, 10.0, skip_vedastro_main_entry_overview=True)
|
||||
misses = []
|
||||
for _index in range(3):
|
||||
started = time.perf_counter()
|
||||
handler._compute_chart_sync(body)
|
||||
misses.append((time.perf_counter() - started) * 1000)
|
||||
os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "900"
|
||||
scratch = Path(tempfile.mkdtemp(prefix="jyotish-hit-time-"))
|
||||
current._api_chart_cache_dir = lambda: scratch
|
||||
cold = handler._compute_chart_sync(body)
|
||||
hits = []
|
||||
for _index in range(3):
|
||||
started = time.perf_counter()
|
||||
hit = handler._compute_chart_sync(body)
|
||||
hits.append((time.perf_counter() - started) * 1000)
|
||||
assert hit.get("runtime_cache", {}).get("cache_hit") is True
|
||||
return {
|
||||
"miss_median_ms": round(statistics.median(misses), 3),
|
||||
"hit_median_ms": round(statistics.median(hits), 3),
|
||||
"cold_has_runtime_cache": isinstance(cold, dict) and "runtime_cache" in cold,
|
||||
}
|
||||
|
||||
|
||||
def main() -> dict:
|
||||
parent_error = None
|
||||
try:
|
||||
parent = _load_parent()
|
||||
except Exception as exc:
|
||||
parent = None
|
||||
parent_error = exc.__class__.__name__
|
||||
rows = _requests()
|
||||
side_a = _hit_rate(parent, rows) if parent is not None else None
|
||||
side_b = _hit_rate(current, rows)
|
||||
report = {
|
||||
"parent": PARENT,
|
||||
"parent_import": parent_error or "ok",
|
||||
"vedastro_sdk": _SDK_PRESENT,
|
||||
"vedastro_network": _NETWORK,
|
||||
"hit_rate_a": side_a,
|
||||
"hit_rate_b": side_b,
|
||||
"artifacts": _artifact_counts(),
|
||||
"negative_cache_files_after_prune": _negative_cache_after_prune(),
|
||||
"prune_10k": _prune_10k(),
|
||||
}
|
||||
if side_a and side_a["hit_rate"]:
|
||||
report["hit_rate_ratio_b_over_a"] = round(side_b["hit_rate"] / side_a["hit_rate"], 3)
|
||||
try:
|
||||
report["files_b"] = _seed_and_prune(current, rows)
|
||||
if parent is not None:
|
||||
report["files_a"] = _seed_and_prune(parent, rows)
|
||||
except Exception as exc:
|
||||
report["files_error"] = exc.__class__.__name__
|
||||
try:
|
||||
report["timing"] = _one_chart_timing()
|
||||
except Exception as exc:
|
||||
report["timing_error"] = exc.__class__.__name__
|
||||
try:
|
||||
report["byte_diff"] = run_byte_diff(parent or current)
|
||||
except Exception as exc:
|
||||
report["byte_diff"] = {"ok": False, "error": exc.__class__.__name__}
|
||||
return report
|
||||
|
||||
|
||||
def _sdk_probe() -> dict:
|
||||
os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "0"
|
||||
os.environ["VEDASTRO_ENABLE_NETWORK"] = "0"
|
||||
os.environ["VEDASTRO_WRITE_AUDIT_ARTIFACTS"] = "0"
|
||||
handler = _handler(current)
|
||||
body = _body(12, 10.0)
|
||||
started = time.perf_counter()
|
||||
try:
|
||||
handler._compute_chart_sync(body)
|
||||
error = None
|
||||
except Exception as exc:
|
||||
error = type(exc).__name__
|
||||
return {
|
||||
"blocked_vedastro": False,
|
||||
"vedastro_sdk": _SDK_PRESENT,
|
||||
"vedastro_network": _NETWORK,
|
||||
"network_enabled": False,
|
||||
"elapsed_s": round(time.perf_counter() - started, 3),
|
||||
"error": error,
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
payload = _sdk_probe() if "--sdk-probe" in sys.argv else main()
|
||||
json.dump(payload, sys.stdout, ensure_ascii=False, indent=2)
|
||||
sys.stdout.write("\n")
|
||||
@@ -340,6 +340,30 @@ def probe_official_rest_health(*, force_refresh: bool = True) -> dict[str, Any]:
|
||||
return result
|
||||
|
||||
|
||||
def _gateway_snapshot_ttl_helpers():
|
||||
try:
|
||||
from vedastro_service_adapter import (
|
||||
_official_full_snapshot_cache_ttl_seconds,
|
||||
_official_full_snapshot_negative_ttl_seconds,
|
||||
)
|
||||
except ModuleNotFoundError:
|
||||
from scripts.vedastro_service_adapter import (
|
||||
_official_full_snapshot_cache_ttl_seconds,
|
||||
_official_full_snapshot_negative_ttl_seconds,
|
||||
)
|
||||
return _official_full_snapshot_cache_ttl_seconds, _official_full_snapshot_negative_ttl_seconds
|
||||
|
||||
|
||||
def _gateway_official_snapshot_ttl_seconds() -> float:
|
||||
positive, _negative = _gateway_snapshot_ttl_helpers()
|
||||
return positive()
|
||||
|
||||
|
||||
def _gateway_official_snapshot_negative_ttl_seconds() -> float:
|
||||
_positive, negative = _gateway_snapshot_ttl_helpers()
|
||||
return negative()
|
||||
|
||||
|
||||
def gateway_status(*, force_refresh: bool = True) -> dict[str, Any]:
|
||||
from scripts.diagnose_vedastro_mode import build_report as build_vedastro_mode_report
|
||||
|
||||
@@ -368,7 +392,8 @@ def gateway_status(*, force_refresh: bool = True) -> dict[str, Any]:
|
||||
),
|
||||
"sdk_version": _sdk_version(),
|
||||
"cache_ttl_seconds": config["cache_ttl_seconds"],
|
||||
"official_full_snapshot_cache_ttl_seconds": _int_env("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", 0),
|
||||
"official_full_snapshot_cache_ttl_seconds": _gateway_official_snapshot_ttl_seconds(),
|
||||
"official_full_snapshot_negative_cache_ttl_seconds": _gateway_official_snapshot_negative_ttl_seconds(),
|
||||
"queue_enabled": config["queue_enabled"],
|
||||
"fail_open_local": config["fail_open_local"],
|
||||
"official_readiness": {
|
||||
|
||||
@@ -425,14 +425,23 @@ def _repo_relative(path: Path) -> str:
|
||||
return str(path)
|
||||
|
||||
|
||||
def _write_artifact(result: dict[str, Any]) -> str:
|
||||
def _audit_artifacts_enabled() -> bool:
|
||||
raw = os.environ.get("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "0").strip().lower()
|
||||
return raw in {"1", "true", "yes", "on"}
|
||||
|
||||
|
||||
def _write_artifact(result: dict[str, Any]) -> str | None:
|
||||
if not _audit_artifacts_enabled():
|
||||
return None
|
||||
metadata = result.get("source_metadata") or {}
|
||||
artifact = _artifact_path(
|
||||
str(metadata.get("operation") or result.get("operation") or "calculation"),
|
||||
str(metadata.get("request_hash") or "no-request-hash"),
|
||||
str(metadata.get("response_hash") or "no-response-hash"),
|
||||
)
|
||||
artifact.write_text(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
|
||||
from api_scratch_retention import write_private_json
|
||||
|
||||
write_private_json(artifact, result)
|
||||
return _repo_relative(artifact)
|
||||
|
||||
|
||||
@@ -486,6 +495,22 @@ def _official_full_snapshot_cache_ttl_seconds() -> float:
|
||||
return 0.0
|
||||
|
||||
|
||||
def _official_full_snapshot_negative_ttl_seconds() -> float:
|
||||
normal = _official_full_snapshot_cache_ttl_seconds()
|
||||
if normal <= 0:
|
||||
return 0.0
|
||||
raw = os.environ.get("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS", "300").strip()
|
||||
try:
|
||||
negative = max(0.0, float(raw)) if raw else 300.0
|
||||
except ValueError:
|
||||
negative = 300.0
|
||||
return min(negative, normal)
|
||||
|
||||
|
||||
def _official_full_snapshot_result_is_success(payload: dict[str, Any]) -> bool:
|
||||
return payload.get("available") is True and payload.get("status") == "ok"
|
||||
|
||||
|
||||
def _free_tier_max_requests() -> int:
|
||||
raw = os.environ.get(FREE_TIER_MAX_REQUESTS_ENV, "").strip()
|
||||
if not raw:
|
||||
@@ -1719,9 +1744,11 @@ def _attach_official_full_snapshot_semantic_cache_metadata(
|
||||
cache_key: str,
|
||||
created_at_unix: float,
|
||||
cache_hit: bool,
|
||||
ttl_seconds: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
payload_copy = json.loads(json.dumps(payload))
|
||||
ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
|
||||
if ttl_seconds is None:
|
||||
ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
|
||||
created_at = datetime.utcfromtimestamp(created_at_unix).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
expires_at = datetime.utcfromtimestamp(created_at_unix + ttl_seconds).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
metadata = dict(payload_copy.get("source_metadata") or {})
|
||||
@@ -1745,9 +1772,15 @@ def _rebind_official_full_snapshot_case_identity(payload: dict[str, Any], case_i
|
||||
return payload_copy
|
||||
|
||||
|
||||
def _semantic_cache_ttl_seconds(record: dict[str, Any]) -> float:
|
||||
# Records written before the negative flag are successes and keep the long TTL.
|
||||
if record.get("negative") is True:
|
||||
return _official_full_snapshot_negative_ttl_seconds()
|
||||
return _official_full_snapshot_cache_ttl_seconds()
|
||||
|
||||
|
||||
def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str) -> dict[str, Any] | None:
|
||||
ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
|
||||
if ttl_seconds <= 0:
|
||||
if _official_full_snapshot_cache_ttl_seconds() <= 0:
|
||||
return None
|
||||
cache_key = _official_full_snapshot_cache_key(case, case_id)
|
||||
cache_path = _official_full_snapshot_cache_path(cache_key)
|
||||
@@ -1763,7 +1796,8 @@ def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: s
|
||||
payload = record.get("payload")
|
||||
if not isinstance(created_at_raw, (int, float)) or not isinstance(payload, dict):
|
||||
return None
|
||||
if time.time() - float(created_at_raw) > ttl_seconds:
|
||||
ttl_seconds = _semantic_cache_ttl_seconds(record)
|
||||
if ttl_seconds <= 0 or time.time() - float(created_at_raw) > ttl_seconds:
|
||||
return None
|
||||
rebound = _rebind_official_full_snapshot_case_identity(payload, case_id)
|
||||
return _attach_official_full_snapshot_semantic_cache_metadata(
|
||||
@@ -1771,28 +1805,37 @@ def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: s
|
||||
cache_key=cache_key,
|
||||
created_at_unix=float(created_at_raw),
|
||||
cache_hit=True,
|
||||
ttl_seconds=ttl_seconds,
|
||||
)
|
||||
|
||||
|
||||
def _store_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
cache_key = _official_full_snapshot_cache_key(case, case_id)
|
||||
created_at_unix = time.time()
|
||||
negative = not _official_full_snapshot_result_is_success(payload)
|
||||
ttl_seconds = (
|
||||
_official_full_snapshot_negative_ttl_seconds()
|
||||
if negative
|
||||
else _official_full_snapshot_cache_ttl_seconds()
|
||||
)
|
||||
payload_copy = _attach_official_full_snapshot_semantic_cache_metadata(
|
||||
payload,
|
||||
cache_key=cache_key,
|
||||
created_at_unix=created_at_unix,
|
||||
cache_hit=False,
|
||||
ttl_seconds=ttl_seconds,
|
||||
)
|
||||
ttl_seconds = _official_full_snapshot_cache_ttl_seconds()
|
||||
if ttl_seconds > 0:
|
||||
record = {
|
||||
"cache_key": cache_key,
|
||||
"created_at": created_at_unix,
|
||||
"payload": payload_copy,
|
||||
}
|
||||
_official_full_snapshot_cache_path(cache_key).write_text(
|
||||
json.dumps(record, ensure_ascii=False, sort_keys=True),
|
||||
encoding="utf-8",
|
||||
from api_scratch_retention import write_private_json
|
||||
|
||||
write_private_json(
|
||||
_official_full_snapshot_cache_path(cache_key),
|
||||
{
|
||||
"cache_key": cache_key,
|
||||
"created_at": created_at_unix,
|
||||
"negative": negative,
|
||||
"payload": payload_copy,
|
||||
},
|
||||
)
|
||||
return payload_copy
|
||||
|
||||
@@ -1882,7 +1925,9 @@ def _store_cached_payload(cache_key: str, payload: dict[str, Any]) -> dict[str,
|
||||
"created_at": created_at_unix,
|
||||
"payload": payload_copy,
|
||||
}
|
||||
cache_path.write_text(json.dumps(record, ensure_ascii=False, sort_keys=True), encoding="utf-8")
|
||||
from api_scratch_retention import write_private_json
|
||||
|
||||
write_private_json(cache_path, record)
|
||||
return payload_copy
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,314 @@
|
||||
"""Retention, chart-cache keys, and VedAstro audit switch."""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
import api_scratch_retention as retention
|
||||
import jyotish_api_server as api
|
||||
import vedastro_service_adapter as adapter
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def _hex(label: str) -> str:
|
||||
return hashlib.sha256(label.encode("utf-8")).hexdigest() + ".json"
|
||||
|
||||
|
||||
def _isolate(monkeypatch, tmp_path: Path) -> None:
|
||||
chart = tmp_path / "chart"
|
||||
jobs = tmp_path / "jobs"
|
||||
chart.mkdir()
|
||||
jobs.mkdir()
|
||||
monkeypatch.setattr(api, "_api_chart_cache_dir", lambda: chart)
|
||||
monkeypatch.setattr(api, "_async_job_dir", lambda scope: jobs)
|
||||
monkeypatch.setattr(api, "_async_job_backend", lambda: "file")
|
||||
monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path / "adapter")
|
||||
monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(tmp_path / "snap"))
|
||||
monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR", str(tmp_path / "neg"))
|
||||
monkeypatch.setenv("VEDASTRO_GATEWAY_QUEUE_DIR", str(tmp_path / "gw"))
|
||||
monkeypatch.setenv("JYOTISH_CACHE_DIR", str(tmp_path / "cache"))
|
||||
monkeypatch.setenv("JYOTISH_API_CHART_CACHE_TTL_SECONDS", "900")
|
||||
monkeypatch.delenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", raising=False)
|
||||
|
||||
|
||||
def _write(path: Path, payload) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(payload), encoding="utf-8")
|
||||
|
||||
|
||||
def test_prune_all_drops_expired_and_old_bad_chart_cache_only(monkeypatch, tmp_path):
|
||||
_isolate(monkeypatch, tmp_path)
|
||||
chart = api._api_chart_cache_dir()
|
||||
now = time.time()
|
||||
fresh = chart / _hex("fresh")
|
||||
expired = chart / _hex("expired")
|
||||
bad_old = chart / _hex("bad-old")
|
||||
bad_young = chart / _hex("bad-young")
|
||||
stranger = chart / "notes.json"
|
||||
_write(fresh, {"cache_key": "a", "created_at": now, "payload": {}})
|
||||
_write(expired, {"cache_key": "b", "created_at": now - 5000, "payload": {}})
|
||||
bad_old.write_text("{", encoding="utf-8")
|
||||
bad_young.write_text("{", encoding="utf-8")
|
||||
stranger.write_text("{}", encoding="utf-8")
|
||||
old = now - (25 * 3600)
|
||||
os.utime(bad_old, (old, old))
|
||||
os.utime(stranger, (old, old))
|
||||
|
||||
result = retention.prune_all(now=now)
|
||||
|
||||
assert fresh.exists()
|
||||
assert bad_young.exists()
|
||||
assert stranger.exists()
|
||||
assert not expired.exists()
|
||||
assert not bad_old.exists()
|
||||
assert result["directories"]["api_chart_cache"]["removed"] == 2
|
||||
assert result["errors"] == 0
|
||||
|
||||
|
||||
def test_job_cleanup_still_removes_only_expired_records(monkeypatch, tmp_path):
|
||||
_isolate(monkeypatch, tmp_path)
|
||||
jobs = api._async_job_dir("any")
|
||||
expired = jobs / "expired.json"
|
||||
active = jobs / "active.json"
|
||||
bad = jobs / "bad.json"
|
||||
expired.write_text(json.dumps({"expires_at_unix": time.time() - 1}), encoding="utf-8")
|
||||
active.write_text(json.dumps({"expires_at_unix": time.time() + 60}), encoding="utf-8")
|
||||
bad.write_text("{", encoding="utf-8")
|
||||
|
||||
result = api.prune_expired_async_jobs()
|
||||
|
||||
assert result["removed"] == 1
|
||||
assert not expired.exists()
|
||||
assert active.exists()
|
||||
assert bad.exists()
|
||||
|
||||
|
||||
def test_interval_zero_does_not_start_a_thread():
|
||||
before = [item.name for item in threading.enumerate()]
|
||||
assert retention.start_periodic_prune(0) is None
|
||||
assert [item.name for item in threading.enumerate()] == before
|
||||
|
||||
|
||||
def test_prune_counts_unlink_errors_without_raising(monkeypatch, tmp_path):
|
||||
_isolate(monkeypatch, tmp_path)
|
||||
chart = api._api_chart_cache_dir()
|
||||
target = chart / _hex("stuck")
|
||||
_write(target, {"created_at": time.time() - 5000, "payload": {}})
|
||||
real_unlink = Path.unlink
|
||||
|
||||
def _boom(self, *args, **kwargs):
|
||||
if self == target:
|
||||
raise OSError("denied")
|
||||
return real_unlink(self, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(Path, "unlink", _boom)
|
||||
result = retention.prune_all(now=time.time())
|
||||
assert result["errors"] > 0
|
||||
assert target.exists()
|
||||
|
||||
|
||||
def _chart_key(body):
|
||||
return api._api_chart_cache_key(api._build_api_chart_cache_payload(body))
|
||||
|
||||
|
||||
def test_numeric_birth_spellings_share_one_chart_cache_key():
|
||||
base = {
|
||||
"year": 1990,
|
||||
"month": 1,
|
||||
"day": 1,
|
||||
"minute": 0,
|
||||
"lat": 28.6139,
|
||||
"lon": 77.2090,
|
||||
"tz": 5.5,
|
||||
"today": "2026-10-14",
|
||||
}
|
||||
keys = {_chart_key(dict(base, hour=hour)) for hour in (12, 12.0, "12")}
|
||||
assert len(keys) == 1
|
||||
assert _chart_key(dict(base, hour=12, second=0)) in keys
|
||||
assert _chart_key(dict(base, hour=12, lat="28.6139")) in keys
|
||||
assert _chart_key(dict(base, hour=12, tz="5.5")) in keys
|
||||
assert _chart_key(dict(base, hour=12, tz="Asia/Kolkata")) not in keys
|
||||
skipped = _chart_key(dict(base, hour=12, skip_vedastro_main_entry_overview=True))
|
||||
assert skipped not in keys
|
||||
assert api._build_api_chart_cache_payload(base)["cache_schema_version"] == 6
|
||||
|
||||
|
||||
def test_audit_artifact_default_off_and_compact_when_on(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path)
|
||||
monkeypatch.delenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", raising=False)
|
||||
manifest = {"requests": [{"section": "chart_core"}], "reference_date": "2026-10-14"}
|
||||
quiet = adapter._normalize_official_full_snapshot_success(
|
||||
"https://example.test/api",
|
||||
manifest,
|
||||
{"chart_core": {"Status": "Pass"}},
|
||||
{"chart_core": "ok"},
|
||||
1,
|
||||
[],
|
||||
)
|
||||
assert quiet["source_metadata"]["artifact_path"] is None
|
||||
assert list(tmp_path.rglob("*.json")) == []
|
||||
|
||||
monkeypatch.setenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "1")
|
||||
written = adapter._normalize_official_full_snapshot_success(
|
||||
"https://example.test/api",
|
||||
manifest,
|
||||
{"chart_core": {"Status": "Pass"}},
|
||||
{"chart_core": "ok"},
|
||||
1,
|
||||
[],
|
||||
)
|
||||
rel = written["source_metadata"]["artifact_path"]
|
||||
path = Path(rel)
|
||||
if not path.is_absolute():
|
||||
path = adapter.ROOT / rel
|
||||
assert path.exists()
|
||||
# Linux reports the mode bit. Windows keeps the file writable and does not surface 0600.
|
||||
if os.name != "nt":
|
||||
assert path.stat().st_mode & 0o777 == 0o600
|
||||
text = path.read_text(encoding="utf-8")
|
||||
assert "\n" not in text
|
||||
assert json.loads(text)["status"] == "ok"
|
||||
|
||||
|
||||
def test_failed_snapshot_expires_before_success(monkeypatch, tmp_path):
|
||||
monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path)
|
||||
monkeypatch.setenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", "604800")
|
||||
monkeypatch.delenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS", raising=False)
|
||||
calls = {"fail": 0, "ok": 0}
|
||||
|
||||
def fake_run(case, case_id="user_chart"):
|
||||
if case.get("day") == 2:
|
||||
calls["fail"] += 1
|
||||
return {"available": False, "status": "timeout", "source_metadata": {}}
|
||||
calls["ok"] += 1
|
||||
return {"available": True, "status": "ok", "source_metadata": {}}
|
||||
|
||||
monkeypatch.setattr(adapter, "_run_official_full_snapshot_case", fake_run)
|
||||
fail_case = {"year": 1990, "month": 1, "day": 2, "hour": 12, "lat": 1.0, "lon": 2.0, "tz": 5.5}
|
||||
ok_case = {"year": 1990, "month": 1, "day": 3, "hour": 12, "lat": 1.0, "lon": 2.0, "tz": 5.5}
|
||||
adapter.run_official_full_snapshot_for_case(fail_case, case_id="fail")
|
||||
adapter.run_official_full_snapshot_for_case(ok_case, case_id="ok")
|
||||
cache_dir = tmp_path / "official_full_snapshot_cache"
|
||||
aged = time.time() - 301
|
||||
for path in cache_dir.glob("*.json"):
|
||||
record = json.loads(path.read_text(encoding="utf-8"))
|
||||
record["created_at"] = aged
|
||||
path.write_text(json.dumps(record), encoding="utf-8")
|
||||
if record.get("negative") is True:
|
||||
assert record["negative"] is True
|
||||
adapter.run_official_full_snapshot_for_case(fail_case, case_id="fail")
|
||||
adapter.run_official_full_snapshot_for_case(ok_case, case_id="ok")
|
||||
assert calls["fail"] == 2
|
||||
assert calls["ok"] == 1
|
||||
|
||||
|
||||
def test_gateway_status_reports_cache_ttl_fallback(monkeypatch):
|
||||
from scripts import vedastro_gateway
|
||||
|
||||
monkeypatch.setenv("JYOTISH_SKIP_LOCAL_ENV", "1")
|
||||
monkeypatch.delenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", raising=False)
|
||||
monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "604800")
|
||||
status = vedastro_gateway.gateway_status()
|
||||
assert status["official_full_snapshot_cache_ttl_seconds"] == 604800
|
||||
assert status["official_full_snapshot_negative_cache_ttl_seconds"] == 300
|
||||
|
||||
|
||||
def _day(offset: int) -> str:
|
||||
return (datetime.now(timezone.utc).date() + timedelta(days=offset)).isoformat()
|
||||
|
||||
|
||||
def test_snapshot_negative_gateway_and_varga_prune_rules(monkeypatch, tmp_path):
|
||||
_isolate(monkeypatch, tmp_path)
|
||||
now = time.time()
|
||||
today = datetime.fromtimestamp(now, timezone.utc).date()
|
||||
snap = tmp_path / "snap"
|
||||
neg = tmp_path / "neg"
|
||||
gw = tmp_path / "gw"
|
||||
varga = tmp_path / "cache" / "varga_segments"
|
||||
keep_ref = (today - timedelta(days=7)).isoformat()
|
||||
drop_ref = (today - timedelta(days=9)).isoformat()
|
||||
_write(snap / _hex("snap-keep"), {"schema": "vedastro_snapshot_cache.v1", "reference_date": keep_ref})
|
||||
_write(snap / _hex("snap-drop"), {"schema": "vedastro_snapshot_cache.v1", "reference_date": drop_ref})
|
||||
_write(snap / "notes.json", {"reference_date": "1990-01-01"})
|
||||
_write(neg / _hex("neg-today"), {"schema": "vedastro_snapshot_negative_cache.v1", "reference_date": today.isoformat()})
|
||||
_write(neg / _hex("neg-yesterday"), {"schema": "vedastro_snapshot_negative_cache.v1", "reference_date": (today - timedelta(days=1)).isoformat()})
|
||||
_write(neg / "stray.json", {"reference_date": "1990-01-01"})
|
||||
job_id = "vgw_" + "ab" * 10
|
||||
old_stamp = datetime.fromtimestamp(now - 90000, timezone.utc).isoformat()
|
||||
fresh_stamp = datetime.fromtimestamp(now, timezone.utc).isoformat()
|
||||
_write(gw / f"{job_id}.json", {"job_id": job_id, "status": "completed", "updated_at": old_stamp})
|
||||
(gw / f"{job_id}.official_raw_response.json").write_text("{}", encoding="utf-8")
|
||||
running = "vgw_" + "cd" * 10
|
||||
_write(gw / f"{running}.json", {"job_id": running, "status": "running", "updated_at": old_stamp})
|
||||
failed = "vgw_" + "ef" * 10
|
||||
_write(gw / f"{failed}.json", {"job_id": failed, "status": "failed", "updated_at": fresh_stamp})
|
||||
(gw / "notes.json").write_text("{}", encoding="utf-8")
|
||||
old_varga = varga / _hex("old-segment")
|
||||
young_varga = varga / _hex("young-segment")
|
||||
_write(old_varga, {"window_key": "x"})
|
||||
_write(young_varga, {"window_key": "y"})
|
||||
_write(varga / "notes.json", {"window_key": "z"})
|
||||
os.utime(old_varga, (now - 8 * 86400, now - 8 * 86400))
|
||||
|
||||
monkeypatch.setenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "0")
|
||||
artifact = tmp_path / "adapter" / "official_full_snapshot-0123456789ab-0123456789ab.json"
|
||||
artifact.parent.mkdir(parents=True, exist_ok=True)
|
||||
artifact.write_text("{}", encoding="utf-8")
|
||||
os.utime(artifact, (now - 25 * 3600, now - 25 * 3600))
|
||||
kept_artifact = tmp_path / "adapter" / "method_catalog_snapshot.json"
|
||||
kept_artifact.write_text("{}", encoding="utf-8")
|
||||
os.utime(kept_artifact, (now - 25 * 3600, now - 25 * 3600))
|
||||
|
||||
retention.prune_all(now=now)
|
||||
|
||||
assert (snap / _hex("snap-keep")).exists()
|
||||
assert not (snap / _hex("snap-drop")).exists()
|
||||
assert (snap / "notes.json").exists()
|
||||
assert (neg / _hex("neg-today")).exists()
|
||||
assert not (neg / _hex("neg-yesterday")).exists()
|
||||
assert (neg / "stray.json").exists()
|
||||
assert not (gw / f"{job_id}.json").exists()
|
||||
assert not (gw / f"{job_id}.official_raw_response.json").exists()
|
||||
assert (gw / f"{running}.json").exists()
|
||||
assert (gw / f"{failed}.json").exists()
|
||||
assert (gw / "notes.json").exists()
|
||||
assert not old_varga.exists()
|
||||
assert young_varga.exists()
|
||||
assert (varga / "notes.json").exists()
|
||||
assert not artifact.exists()
|
||||
assert kept_artifact.exists()
|
||||
|
||||
|
||||
def _scratch_local_names() -> set[str]:
|
||||
names: set[str] = set()
|
||||
slash = re.compile(r"scratch/local/([A-Za-z0-9_.]+)")
|
||||
joined = re.compile(
|
||||
r"""['\"]scratch['\"]\s*/\s*['\"]local['\"]\s*/\s*(?:f)?['\"]([A-Za-z0-9_.]+)['\"]"""
|
||||
)
|
||||
after_local = re.compile(
|
||||
r"""['\"]local['\"]\s*\)*\s*/\s*['\"]([A-Za-z0-9_.]+)['\"]"""
|
||||
)
|
||||
for path in (ROOT / "scripts").rglob("*.py"):
|
||||
text = path.read_text(encoding="utf-8")
|
||||
names.update(slash.findall(text))
|
||||
names.update(joined.findall(text))
|
||||
names.update(after_local.findall(text))
|
||||
return names
|
||||
|
||||
|
||||
def test_new_scratch_local_dirs_are_registered_or_whitelisted():
|
||||
allowed = set(retention.REGISTERED_DIR_NAMES) | set(retention.OFFLINE_SCRATCH_NAMES)
|
||||
found = _scratch_local_names()
|
||||
unexpected = sorted(found - allowed)
|
||||
assert unexpected == [], unexpected
|
||||
assert "api_chart_cache" in found
|
||||
assert "consult_full_data_cache" in retention.OFFLINE_SCRATCH_NAMES
|
||||
@@ -769,6 +769,10 @@ def test_vedastro_range_scan_records_hashes_and_artifact_path() -> None:
|
||||
env = os.environ.copy()
|
||||
env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/vedastro"
|
||||
env["VEDASTRO_ENABLE_NETWORK"] = "1"
|
||||
# 原值:沿用进程环境,没有 VEDASTRO_WRITE_AUDIT_ARTIFACTS。
|
||||
# 新值:显式设为 1。
|
||||
# 原因:审计落盘默认关闭后,这条测试仍要断言文件存在,所以打开开关。这是测试前置,不是放宽断言。
|
||||
env["VEDASTRO_WRITE_AUDIT_ARTIFACTS"] = "1"
|
||||
completed = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
|
||||
Reference in New Issue
Block a user