Simulation Database API Reference

Simulation archive database — schema and queue interfaces.

This module defines the v2 archive layout described in DESIGN.md: manifest.json + index.jsonl + per-sim status.json/summary.json under sims/<name>/, with frozen code under code/, and an explicit split between RequestQueue (decides what to submit) and RunQueue (actually runs the generator).

The existing classes in BaseManager.py / CondorManager.py / SlurmManager.py keep working unchanged; this module is additive. Real condor/slurm queue implementations should land here as small subclasses of RequestQueue / RunQueue.

Status:

Manifest, Index, StatusRecord, code-freeze: implemented.
LocalRequestQueue + LocalRunQueue: implemented (no schedd; runs the frozen generator inline). Used by examples and tests.
DualCondorRequestQueue + DualCondorRunQueue: implemented. The run queue writes per-(sim, level) submit descriptions, assembles a chained DAG, and dispatches via condor_submit_dag (with -name <run_pool> for cross-pool). poll() queries the run pool’s schedd via the cached _htcondor_module from CondorManager (htcondor / htcondor2). Output-on-disk drives final ‘complete’ transitions via Archive.refresh_status_from_disk.
SlurmRunQueue: not yet provided (stubbed out at the design level only). Build on simple_slurm or pyslurmutils; mirror the condor implementation.
make_queues_from_manifest(archive): instantiates queues from the manifest’s request_queue / run_queue config. Used by cli/request_sim.py –ensure to attach queues automatically.

Bases: object

Thin facade over the on-disk layout. Owns the FSM; queues report state, this class applies it. Construction either creates a new archive (when manifest is given) or rehydrates an existing one (when only base_location is given).

expected_output(sim_name: str, level: int) → Tuple[str, str]: Return (output_basename, absolute_remap_target) for a (sim, level) worker. Use the basename in transfer_output_files and the pair in transfer_output_remaps so condor places the worker’s output at the canonical archive location on the submit node.

find_existing(params: Any) → str | None: Return the name of an existing sim whose params satisfy same_q(params, existing.params), else None. O(1) average via the lookup_key bucket; O(K) worst case where K is the bucket size (typically 1).

get_resources(name: str) → Dict[str, Any]

get_status(name: str) → str | None

load_generator() → Callable[[...], Any]

load_summarizer() → Callable[[...], Any] | None

rebuild_index() → int: Reconstruct index.jsonl from per-sim status.json files. Useful if the index is corrupted, lost, or out of sync (e.g. after manual file ops). Returns the count of rows written. Re-runs the summarizer would be a separate pass (resummarize_all).

refine(name: str, target_level: int) → bool: Explicit refinement request. Bumps the target_level if needed and (when the sim already had output) transitions it to ‘refine_ready’. Returns True iff a bump occurred.

refresh_status_from_disk() → Dict[str, str]

Sweep every sim and reconcile its levels[] history with the files actually present under sims/<name>/. New level_<N>.json files (e.g. dropped in by a condor worker via transfer_output_remaps) are absorbed; sims promote to ‘complete’ when current_level reaches target_level, or to ‘refine_ready’ if they’re not done.

Returns a dict mapping sim_name -> new_status for every sim whose status changed. Safe to call repeatedly (idempotent when no new outputs have arrived).

register(params: Any, target_level: int = 1, name: str | None = None) → str

Idempotent under same_q + level. Behavior:

sim does not exist: register fresh with target_level=N

sim exists, current_level >= N: return name, no work

sim exists, current_level < N: bump target_level to max(existing, N), transition to ‘refine_ready’, return name

Returns the (existing or newly allocated) sim_name in all cases. Multi-writer-safe: holds the archive lock for the duration.

resummarize_all(*, only_complete: bool = False) → Dict[str, str]

Re-load the manifest’s summarizer and re-run it for every sim with at least one completed level. Used after the summarizer’s source has been updated (a new freeze + manifest rewrite, or an in-place edit of code/summarizer.py for archives whose summarizer is allowed to evolve).

Returns {sim_name: ‘ok’ | ‘no-summarizer’ | ‘no-levels’ | ‘error: <msg>’}. With only_complete=True skips sims whose status isn’t ‘complete’.

set_resources(name: str, request_memory: int | None = None, request_disk: str | None = None, extra_condor_cmds: Dict[str, str] | None = None) → None: Per-sim overrides on top of the queue’s defaults. Only the keys you pass are updated; pass None to leave a field unchanged. The queue’s build_worker reads these and merges them on top of its own defaults at submit time.

sim_dir(name: str) → Path

simulations_iter_names() → Iterable[str]

transfer_input_files_for(sim_name: str, level: int) → List[str]

Files a worker needs on the execute host to compute (sim, level).

Suitable as the value of condor’s transfer_input_files. Paths are returned as filesystem paths on the submit side; condor flattens basenames into the worker sandbox cwd.

Includes:

<archive>/code — the whole frozen code directory
<archive>/sims/<name>/params.json
<archive>/sims/<name>/level_1.json … level_<level-1>.json (whichever are present on disk)

transition(name: str, new_status: str, **fields: Any) → None

unstick(name: str, bump_memory: bool = False, bump_factor: float = 1.5) → None: Clear stuck, transition to refine_ready (if has prior levels) or ready (if not). With bump_memory=True, also multiplies the per-sim request_memory by bump_factor so the next attempt asks for more headroom.

unstick_all(bump_memory: bool = False, bump_factor: float = 1.5) → List[str]

update_summary(name: str, summary: Dict[str, Any]) → None

verify() → Dict[str, Any]: Cross-check index.jsonl, per-sim status.json, and on-disk files. Returns a structured report — meant to be inspected by an operator before acting on it. Does NOT mutate; pair with rebuild_index if the report shows drift you want to repair.

with_status(status: str) → List[str]

worker_bootstrap_script() → str

Return a self-contained Python bootstrap script (as a string) that the run pool’s executable should be set to. The script expects argv:

–sim-name <n> –level <N> [–code-dir <path>] (default: ‘code’ relative to cwd;

slurm passes the archive’s absolute code/ path here)

[–prev-levels FILE […]]

Reads ./params.json from cwd, writes ./level_<N>.json to cwd. Used by both DualCondorRunQueue (sandbox cwd, files flattened) and SlurmRunQueue (cwd = sim_dir on shared FS).

RIFT.simulation_manager.database.DEFAULT_OOM_HOLD_CODES = (34, 26)

Hold codes this class treats as “the job ran out of memory”, and the attribute that rations retries. Both are DEFAULTS, not facts: what a hold code means is a property of the site, not of HTCondor. 34 is the unambiguous memory code; 26 is SystemPolicy, which means whatever the site’s SYSTEM_PERIODIC_HOLD expressions say it means – on the LIGO clusters this policy was written for that is usually memory, and on an OSG access point it is as likely to be an anti-thrash limiter whose precondition is a high NumJobStarts. Sites that differ pass oom_hold_codes / oom_hold_subcode_exclusions / oom_retry_counter rather than editing this.

Deliberately NOT recorded here: which sites differ, and how. That belongs in whatever inventory the operator already keeps about their own infrastructure. A table of site facts in shared code is stale the day after it is written and wrong for everyone it does not name.

class RIFT.simulation_manager.database.DualCondorRequestQueue(run_queue: DualCondorRunQueue | None = None, request_pool: str | None = None, run_pool: str | None = None, run_collector: str | None = None, **submit_kwargs: Any)

Bases: RequestQueue

Runs on the request condor pool. Its job is orchestration: pick the sims that need work, ask the run queue to actually dispatch them, and poll for completion.

In the simplest deployment the request queue is essentially a thin wrapper over the run queue — the planner that decides what to ask for lives upstream (e.g. the user’s hyperpipeline DAG calling Archive.register(params, target_level=N) per node).

Configuration in the manifest’s request_queue.extra:: request_pool : str # informational; the schedd this DAG runs on run_pool : str # passthrough to the run queue run_collector : str # collector host for htcondor.Schedd lookups accounting_group / accounting_group_user

kind: str = 'condor'

poll(archive: Archive) → Dict[str, str]

submit_pending(archive: Archive) → List[str]

class RIFT.simulation_manager.database.DualCondorRunQueue(run_pool: str | None = None, run_collector: str | None = None, request_memory: int = 4096, request_disk: str = '4G', accounting_group: str | None = None, accounting_group_user: str | None = None, getenv: str | None = None, use_singularity: bool = False, singularity_image: str | None = None, extra_condor_cmds: Dict[str, str] | None = None, extra_transfer_input_files: Sequence[str] | None = None, extra_transfer_output_files: Sequence[str] | None = None, auto_release_on_oom: bool = True, extra_periodic_release: str | None = None, oom_hold_codes: Sequence[int] | None = None, oom_hold_subcode_exclusions: Mapping[int, Sequence[int]] | None = None, oom_retry_counter: str | None = None, oom_max_retries: int = 5, oom_memory_factor: float = 1.5, subdag_factory: Callable[[Any, str, int], str] | None = None, submit_mode: str = 'submit', *, container_image: str | None = None, when_to_transfer_output: str | None = None, **submit_kwargs: Any)

Bases: RunQueue

Run-pool queue: ferries per-(sim, level) work to condor execute hosts and tracks completion via the schedd.

Each (sim, level) is one condor job. Levels for the same sim are chained as DAG parents/children so an accumulating generator sees its prior levels’ outputs already on disk before its job starts.

Configuration:

run_poolstr – target schedd (-name <run_pool> on: condor_submit_dag for cross-pool). None = local schedd.
run_collectorstr – collector host for cross-pool: htcondor.Schedd(<collector>) queries in poll(). None = local.

accounting_group – defaults to env LIGO_ACCOUNTING accounting_group_user – defaults to env LIGO_USER_NAME

Both follow the standard LIGO/IGWN convention; the matching env-var fallback matches the legacy CondorManager behaviour so existing submit hosts work unchanged.

request_memory : int (MB), default 4096 request_disk : str (e.g. ‘4G’), default ‘4G’ getenv : str – value of condor ‘getenv’ command.

Default precedence: constructor kwarg > $RIFT_GETENV env > safe allowlist (‘LD_LIBRARY_PATH,PATH,PYTHONPATH,*RIFT*, LIBRARY_PATH’). NOTE: getenv = True is blocked by many sites (CIT among them); the allowlist is the OSG-blessed alternative. Pass getenv=’True’ explicitly only on sites that allow it.

container_imagestr – KEYWORD-ONLY. Image reference for: HTCondor’s container universe (osdf://, docker://, a local .sif). Setting it switches the job from vanilla to universe = container. Mutually exclusive with use_singularity, which is the legacy +SingularityImage form.
when_to_transfer_output: str – KEYWORD-ONLY. One of ON_EXIT: (default), ON_EXIT_OR_EVICT, ON_SUCCESS, NEVER. ON_EXIT discards the sandbox on eviction, so a preemptable pool loses whatever the job had already written. Like container_image and the transfer extras, this is emitted by build_worker and so is refused together with subdag_factory: a sub-DAG writes its own submit descriptions, and its nodes would transfer at the default instead.

use_singularity : bool singularity_image: str – required if use_singularity=True oom_hold_codes : seq – hold codes this site reports when a

job runs out of memory. Default DEFAULT_OOM_HOLD_CODES = (34, 26). 34 is unambiguous; 26 is SystemPolicy and means whatever the site’s SYSTEM_PERIODIC_HOLD expressions say, which elsewhere may be an anti-thrash limiter rather than memory.

oom_hold_subcode_exclusions: {code: [subcode, …]} – sub-codes: to carve out of a code above. Needed because every SYSTEM_PERIODIC_HOLD at a site reports one hold code and only the sub-code separates “over memory” from “restarted too many times”. A sub-code keyed on a code not listed in oom_hold_codes is refused rather than ignored.
oom_retry_counter: str – ClassAd expression rationing the: retries and scaling the bump. Default DEFAULT_OOM_RETRY_COUNTER = “NumJobStarts”. NumHolds is the other obvious choice and is not better everywhere: it counts holds of every kind, including transfer failures that increment it without the job ever running.
extra_periodic_release: str – a ClassAd expression OR’d into: periodic_release alongside the OOM policy, for sites that hold jobs for reasons this class does not model. While auto_release_on_oom is on, the term is scoped away from whatever codes oom_hold_codes names, so oom_max_retries stays a real cap and the term governs every other code. With the OOM policy off it governs all of them. Setting periodic_release through extra_condor_cmds is refused – it replaced the whole expression and dropped the memory handling with it.
extra_condor_cmds: dict – additional key = value lines: appended verbatim to the submit description (e.g. +DESIRED_SITES, +UNDESIRED_SITES for OSG site selection, requirements clauses).
extra_transfer_input_files: list – extra entries APPENDED to: transfer_input_files for every job. Intended for bulk inputs addressed by URL (osdf://, http://) so they are fetched from a cache instead of staged through the submit host’s spool. Setting transfer_input_files via extra_condor_cmds would instead replace the archive’s own entries and strip the frozen code/ directory, leaving the worker nothing to run.
extra_transfer_output_files: list – products to bring BACK: beyond the level_<N>.json marker, named relative to the job sandbox. {level} and {sim_name} are substituted, so e.g. “level_{level}” returns a per-level output directory. Each is remapped to the same relative path under sims/<name>/. transfer_output_files is explicit, so without this HTCondor returns only the marker and everything else the worker produced dies with the sandbox — the job completes having discarded its own results.

The defaults above also apply when DualCondorRunQueue is instantiated via make_queues_from_manifest() — keys absent from run_queue.extra in the manifest fall back to the constructor defaults, which in turn fall back to the env vars listed above.

build_worker(archive: Archive, sim_name: str, level: int = 1) → str

Write the per-(sim, level) condor submit description and return its absolute path. Idempotent: re-running it overwrites.

Per-sim resource overrides (set via Archive.set_resources) merge on top of the queue’s defaults: per-sim values for request_memory / request_disk / extra_condor_cmds win where set.

property container_image: str

property extra_periodic_release: str

property extra_transfer_input_files: Tuple[str, ...]

property extra_transfer_output_files: Tuple[str, ...]

kind: str = 'condor'

property oom_hold_codes: Tuple[int, ...]

property oom_hold_subcode_exclusions: Mapping[int, Tuple[int, ...]]

property oom_retry_counter: str

poll(archive: Archive, sim_names: Iterable[str]) → Dict[str, str]

Query the run pool’s schedd; sims with at least one job still in the queue go to ‘running’; sims that previously had jobs but now don’t, with their target levels’ output files present, are reconciled by Archive.refresh_status_from_disk (called by the request queue right after this returns).

Works with htcondor or htcondor2 via the cached _htcondor_module from CondorManager. If neither binding is available, returns the currently-recorded statuses without modification.

submit(archive: Archive, sim_names: Iterable[str]) → List[Tuple[str, str]]

For each sim, build its per-level work units (vanilla-universe .sub via build_worker, OR a sub-DAG via subdag_factory) and assemble a wrapper DAG with PARENT/CHILD edges between the levels of each sim. Behavior of the dispatch step depends on self.submit_mode:

“submit” (default): invoke condor_submit_dag and return [(sim, cluster_id), …].

“embed”: write the wrapper DAG and return [(sim, “”), …] without dispatching. The wrapper DAG path is recorded on self.last_wrapper_dag_path so a parent workflow can pick it up via SUBDAG EXTERNAL <id> <wrapper_path>.

When self.subdag_factory is set, each per-(sim, level) node in the wrapper DAG is SUBDAG EXTERNAL rather than JOB. This lets backends whose work unit is itself a DAG (e.g. GW PE via util_RIFT_pseudo_pipe) compose cleanly.

property when_to_transfer_output: str

class RIFT.simulation_manager.database.Index(base: str | Path)

Bases: object

One JSON object per line, one line per sim. The canonical, cheap view of the archive. Status updates and registrations rewrite the file under a simple file lock (not implemented yet — single-writer assumption is documented for now).

FILENAME = 'index.jsonl'

all() → List[Dict[str, Any]]

by_name(name: str) → Dict[str, Any] | None

property path: Path

remove(name: str) → None

upsert(row: Dict[str, Any]) → None

with_status(status: str) → List[Dict[str, Any]]

class RIFT.simulation_manager.database.LocalRequestQueue(run_queue: LocalRunQueue)

Bases: RequestQueue

Trivial pass-through: every sim that needs work (‘ready’ or ‘refine_ready’) is immediately handed to the run queue.

kind: str = 'local'

poll(archive: Archive) → Dict[str, str]

submit_pending(archive: Archive) → List[str]

class RIFT.simulation_manager.database.LocalRunQueue

Bases: RunQueue

Runs the frozen generator inline in the current process. Computes every missing level (current_level + 1 … target_level) for each submitted sim. Useful for end-to-end tests without a cluster.

build_worker(archive: Archive, sim_name: str) → str

kind: str = 'local'

poll(archive: Archive, sim_names: Iterable[str]) → Dict[str, str]

submit(archive: Archive, sim_names: Iterable[str]) → List[Tuple[str, str]]

class RIFT.simulation_manager.database.Manifest(data: Dict[str, Any])

Bases: object

archive-level metadata stored at <base>/manifest.json.

FILENAME = 'manifest.json'

classmethod new(name: str, request_queue_kind: str, run_queue_kind: str, generator_entrypoint: str = 'generator:run', summarizer_entrypoint: str | None = None, same_q_entrypoint: str | None = None, lookup_key_entrypoint: str | None = None, params_schema: Dict[str, Any] | None = None, summary_schema: Dict[str, Any] | None = None, rift_version: str | None = None, request_queue_extra: Dict[str, Any] | None = None, run_queue_extra: Dict[str, Any] | None = None) → Manifest

classmethod read(base: str | Path) → Manifest

write(base: str | Path) → None

class RIFT.simulation_manager.database.RequestQueue

Bases: object

Decides which sims should next be sent to the run queue, and tracks their state in the request system. Subclasses set kind.

kind: str = 'abstract'

poll(archive: Archive) → Dict[str, str]

submit_pending(archive: Archive) → List[str]

class RIFT.simulation_manager.database.RunQueue

Bases: object

Actually runs sims (writes their output).

build_worker(archive: Archive, sim_name: str) → str

kind: str = 'abstract'

poll(archive: Archive, sim_names: Iterable[str]) → Dict[str, str]

submit(archive: Archive, sim_names: Iterable[str]) → List[Tuple[str, str]]

class RIFT.simulation_manager.database.SlurmRequestQueue(run_queue: SlurmRunQueue | None = None, **submit_kwargs: Any)

Bases: RequestQueue

Slurm-side orchestrator. Identifies pending sims and delegates submission to SlurmRunQueue.

kind: str = 'slurm'

poll(archive: Archive) → Dict[str, str]

submit_pending(archive: Archive) → List[str]

class RIFT.simulation_manager.database.SlurmRunQueue(partition: str | None = None, time_limit: str = '01:00:00', nodes: int = 1, ntasks: int = 1, cpus_per_task: int = 1, request_memory: int = 4096, request_disk: str | None = None, account: str | None = None, qos: str | None = None, partition_extra: Dict[str, str] | None = None, sbatch_path: str = 'sbatch', squeue_path: str = 'squeue', sacct_path: str = 'sacct', python_executable: str = 'python3', prelude: str = '', **submit_kwargs: Any)

Bases: RunQueue

Per-(sim, level) worker is one sbatch script. Levels for the same sim are chained via –dependency=afterok:<previous-jobid>.

Configuration:

partition : str – required: SLURM partition / queue time_limit : str – e.g. ‘02:00:00’ (default ‘01:00:00’) nodes : int – default 1 ntasks : int – default 1 cpus_per_task : int – default 1 request_memory : int (MB), default 4096 request_disk : str – ignored by slurm directly; informational account : str – defaults to env SLURM_ACCOUNT qos : str – defaults to env SLURM_QOS partition_extra : dict – additional #SBATCH –key=value lines

emitted verbatim

sbatch_path : str – default ‘sbatch’ squeue_path : str – default ‘squeue’ sacct_path : str – default ‘sacct’ python_executable : str – default ‘python3’; what to invoke the

bootstrap with

preludestr – shell snippet inserted before the: bootstrap (module load python, source /path/to/env/bin/activate, singularity exec … python3 …)

Compared to DualCondorRunQueue, this queue does NOT bake in an auto-OOM-release mechanism. Slurm’s standard tool for that is the job’s –requeue policy combined with operator-driven scontrol update jobid=… ReqMem=…; users who want automated retry should use Archive.unstick(name, bump_memory=True) and re-submit, or wrap the sbatch in a re-submit shell harness.

build_worker(archive: Archive, sim_name: str, level: int = 1) → str: Write the per-(sim, level) sbatch shell script and return its absolute path. Idempotent.

kind: str = 'slurm'

poll(archive: Archive, sim_names: Iterable[str]) → Dict[str, str]: Use squeue to find jobs still in the queue, then sacct to confirm completion/failure for ones that have left. Updates archive statuses; output-on-disk is the authoritative completion signal (applied by the request queue’s poll via Archive.refresh_status_from_disk).

submit(archive: Archive, sim_names: Iterable[str]) → List[Tuple[str, str]]: Build per-(sim, level) sbatch scripts and submit them with –dependency=afterok chains within each sim. Returns [(sim_name, last_level_jobid_or_empty), …].

class RIFT.simulation_manager.database.StatusRecord(data: Dict[str, Any])

Bases: object

sims/<name>/status.json. The full per-sim story; the index is a projection of this. Mutations go through transition() so the history log and the FSM stay consistent.

FILENAME = 'status.json'

append_level(level: int, output_path: str) → None: Record a successful level computation. Updates current_level and the levels[] list. Caller should subsequently transition to ‘complete’ or ‘refine_ready’ as appropriate.

bump_target(target_level: int) → bool: Raise target_level if target_level is higher than the current target. Returns True iff a bump occurred.

needs_more_work() → bool

classmethod new(name: str, params: Any, target_level: int = 1) → StatusRecord

classmethod read(sim_dir: str | Path) → StatusRecord

transition(new_status: str, **fields: Any) → None

write(sim_dir: str | Path) → None

RIFT.simulation_manager.database.WHEN_TO_TRANSFER_OUTPUT = ('ON_EXIT', 'ON_EXIT_OR_EVICT', 'ON_SUCCESS'): HTCondor’s when_to_transfer_output vocabulary, as the JDL defines it. Three values, not four: NEVER is NOT one of them, and condor does not say so – measured against condor 25.13.1, NEVER submits with rc=0 and materialises as WhenToTransferOutput=”ON_EXIT”, silently. A caller setting it to suppress transfer gets transfer. BANANA is rejected loudly, so listing NEVER blessed the one value condor discards while catching nothing condor would not have caught.

RIFT.simulation_manager.database.freeze_code(spec: Callable[[...], Any] | str | PathLike | Dict[str, Any], code_dir: str | Path, target_filename: str = 'generator.py') → str

Snapshot the generator (or summarizer) source into <code_dir>/.

Returns the entrypoint string (“module:callable”) that should be stored in the manifest. Three input shapes are accepted:

callable: inspect.getsource() is captured into a single-file module named after target_filename. The function must be self-contained (no closure captures, no module-relative imports beyond the standard library and explicitly-listed deps).

path: a .py file is copied verbatim.

dict: {“module_path”: …, “entrypoint”: “mod:fn”,
“extra_files”: […]}. All listed files are copied; the

named module becomes the canonical generator.

RIFT.simulation_manager.database.load_entrypoint(code_dir: str | Path, entrypoint: str) → Callable[[...], Any]: Resolve a ‘module:callable’ entrypoint against <code_dir>/. Used by workers (and by LocalRunQueue for the no-schedd path).

RIFT.simulation_manager.database.make_queues_from_manifest(archive: Archive) → Tuple[RequestQueue, RunQueue]

Instantiate (request_queue, run_queue) from the archive’s manifest. The run queue’s extra dict is passed as keyword arguments; the request queue gets the run_queue plumbed in.

Used by the request_sim CLI’s –ensure mode so the user doesn’t have to wire queues by hand.