diff --git a/docs/config.yaml.template b/docs/config.yaml.template index 6b005cf1..24b0fa73 100644 --- a/docs/config.yaml.template +++ b/docs/config.yaml.template @@ -137,6 +137,15 @@ session_cookie_secure: true # extra_paths: # Paths appended to $PATH in every job script # - /opt/nextflow/bin # and used when verifying tool requirements # - /opt/pixi/bin +# worker_env_passthrough: # Extra env var names (exact) or prefixes +# - MYSITE_LICENSE_SERVER # (ending in '_') to pass to per-user workers, +# - MYSITE_ # on top of the built-in allowlist. The worker +# # env is an allowlist so no server secret leaks +# # to the user; add only non-secret vars your +# # scheduler or tools need. FGC_* is never passed. +# unknown_timeout_hours: 24 # Give up polling a job stuck in UNKNOWN after +# # this many hours and mark it FAILED (the +# # scheduler can no longer report it). 0 disables. # Cluster configuration # Mirrors py-cluster-api ClusterConfig - all fields are optional diff --git a/fileglancer/alembic/versions/a7e2f9d31c04_add_status_updated_at_to_jobs.py b/fileglancer/alembic/versions/a7e2f9d31c04_add_status_updated_at_to_jobs.py new file mode 100644 index 00000000..3a2680e8 --- /dev/null +++ b/fileglancer/alembic/versions/a7e2f9d31c04_add_status_updated_at_to_jobs.py @@ -0,0 +1,24 @@ +"""add status_updated_at column to jobs + +Revision ID: a7e2f9d31c04 +Revises: f4a1d8c62e97 +Create Date: 2026-07-04 00:00:00.000000 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'a7e2f9d31c04' +down_revision = 'f4a1d8c62e97' +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column('jobs', sa.Column('status_updated_at', sa.DateTime(), nullable=True)) + + +def downgrade() -> None: + op.drop_column('jobs', 'status_updated_at') diff --git a/fileglancer/alembic/versions/f4a1d8c62e97_add_commit_pinning_to_apps_and_jobs.py b/fileglancer/alembic/versions/f4a1d8c62e97_add_commit_pinning_to_apps_and_jobs.py new file mode 100644 index 00000000..8ec73b4a --- /dev/null +++ b/fileglancer/alembic/versions/f4a1d8c62e97_add_commit_pinning_to_apps_and_jobs.py @@ -0,0 +1,56 @@ +"""add commit pinning to user_apps and jobs + +Revision ID: f4a1d8c62e97 +Revises: c1f9a4e7b2d8 +Create Date: 2026-07-02 00:00:00.000000 + +user_apps.commit_sha pins each app to the exact commit it was added or last +updated at; jobs run from an immutable per-SHA snapshot of that commit rather +than the mutable branch clone. code_commit_sha is the equivalent pin for a +manifest's separate code repo (repo_url), when one is declared. +jobs.commit_sha records the commit whose code the job actually executed, and +jobs.code_repo_url the repo that commit belongs to when it isn't the app repo. + +All columns are nullable: legacy rows have no pin and are backfilled the next +time the app is launched or updated. +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = 'f4a1d8c62e97' +down_revision = 'c1f9a4e7b2d8' +branch_labels = None +depends_on = None + + +def _missing(table: str, column: str) -> bool: + inspector = sa.inspect(op.get_bind()) + return column not in {c["name"] for c in inspector.get_columns(table)} + + +def upgrade() -> None: + # Guarded adds: this revision briefly existed without jobs.code_repo_url, + # so a dev database migrated during that window is stamped at this head + # while missing the column. Re-running after `alembic stamp c1f9a4e7b2d8` + # converges any such state without erroring on the columns that do exist. + if _missing('user_apps', 'commit_sha'): + op.add_column('user_apps', sa.Column('commit_sha', sa.String(), nullable=True)) + if _missing('user_apps', 'code_commit_sha'): + op.add_column('user_apps', sa.Column('code_commit_sha', sa.String(), nullable=True)) + if _missing('jobs', 'commit_sha'): + op.add_column('jobs', sa.Column('commit_sha', sa.String(), nullable=True)) + if _missing('jobs', 'code_repo_url'): + op.add_column('jobs', sa.Column('code_repo_url', sa.String(), nullable=True)) + + +def downgrade() -> None: + if not _missing('jobs', 'code_repo_url'): + op.drop_column('jobs', 'code_repo_url') + if not _missing('jobs', 'commit_sha'): + op.drop_column('jobs', 'commit_sha') + if not _missing('user_apps', 'code_commit_sha'): + op.drop_column('user_apps', 'code_commit_sha') + if not _missing('user_apps', 'commit_sha'): + op.drop_column('user_apps', 'commit_sha') diff --git a/fileglancer/apps/__init__.py b/fileglancer/apps/__init__.py index ad92584b..45f0ce0b 100644 --- a/fileglancer/apps/__init__.py +++ b/fileglancer/apps/__init__.py @@ -5,17 +5,23 @@ _ensure_repo_cache, clone_url_for_stored_app, discover_app_manifests, + ensure_repo_snapshot, fetch_app_manifest, + gc_repo_snapshots, get_app_branch, + get_remote_head, + get_remote_heads, canonical_app_url, get_or_load_manifest, refresh_cached_manifest, set_worker_exec, + validate_commit_sha, ) from fileglancer.apps.command import ( # noqa: F401 _TOOL_REGISTRY, build_command, build_requirements_check, + collect_creatable_dirs, collect_path_parameters, expand_user_path, merge_requirements, @@ -24,7 +30,10 @@ ) from fileglancer.apps.jobs import ( # noqa: F401 _build_container_script, + _build_service_url_publisher, + _container_bind_paths, _container_sif_name, + _SERVICE_PORT_HELPER, cancel_job, start_job_monitor, stop_job_monitor, @@ -34,4 +43,5 @@ get_job_file_content, get_job_file_paths, get_service_url, + get_service_phase, ) diff --git a/fileglancer/apps/command.py b/fileglancer/apps/command.py index d7a161eb..86bd4294 100644 --- a/fileglancer/apps/command.py +++ b/fileglancer/apps/command.py @@ -268,19 +268,26 @@ def validate_path_for_shell(path_value: str) -> str | None: return None -def validate_path_in_filestore(path_value: str, fsps: list, check_access: bool = True) -> str | None: +def validate_path_in_filestore(path_value: str, fsps: list, check_access: bool = True, + expected_type: str | None = None) -> str | None: """Validate a path within an allowed file share. Always performs syntax checks and confirms the path resolves within an allowed file share; both checks are euid-independent. When check_access is True, additionally verifies the path exists and is readable. - The exists/readable check reflects the *calling process's* identity, so it - is only meaningful when this runs as the target user (i.e. in the setuid - worker). Callers running on the root server must pass check_access=False — - otherwise validation reflects root's access, not the user's (false-pass on - local FS where root bypasses perms, false-reject on root-squash NFS) — and - defer the access check to the worker (see submit_job). + When expected_type is 'file' or 'directory' and the path exists, its type + must match — a folder pasted into a file parameter (or vice versa) is + rejected here rather than failing at job runtime. A missing path is not + type-checked (an exists=false output has no type yet). + + The exists/readable/type checks reflect the *calling process's* identity, + so they are only meaningful when this runs as the target user (i.e. in the + setuid worker). Callers running on the root server must pass + check_access=False — otherwise validation reflects root's access, not the + user's (false-pass on local FS where root bypasses perms, false-reject on + root-squash NFS) — and defer the access check to the worker (see + submit_job). Returns an error message string if invalid, or None if valid. """ @@ -304,14 +311,23 @@ def validate_path_in_filestore(path_value: str, fsps: list, check_access: bool = if result is None: return "Path is not within an allowed file share" - if not check_access: - return None - fsp, subpath = result - from fileglancer.filestore import Filestore - filestore = Filestore(fsp) - return filestore.validate_path(subpath) + if check_access: + from fileglancer.filestore import Filestore + filestore = Filestore(fsp) + error = filestore.validate_path(subpath) + if error: + return error + + if expected_type in ("file", "directory") and os.path.exists(expanded): + is_dir = os.path.isdir(expanded) + if expected_type == "file" and is_dir: + return "Path is a folder, but a file is required" + if expected_type == "directory" and not is_dir: + return "Path is a file, but a folder is required" + + return None # --- Command Building --- @@ -375,7 +391,8 @@ def _validate_parameter_value(param: AppParameter, value, session=None, username str_val = expand_user_path(str_val, username) if session is not None: fsps = db.get_file_share_paths(session) - error = validate_path_in_filestore(str_val, fsps, check_access=check_access) + error = validate_path_in_filestore(str_val, fsps, check_access=check_access, + expected_type=param.type) else: error = validate_path_for_shell(str_val) if error: @@ -399,6 +416,15 @@ def _flatten_param_items(items) -> list: return result +def _is_nextflow_run_command(command: str) -> bool: + """Return true when the base command appears to be ``nextflow run``.""" + try: + argv = shlex.split(command) + except ValueError: + argv = command.strip().split() + return len(argv) >= 2 and argv[0] == "nextflow" and argv[1] == "run" + + def build_command(entry_point: AppEntryPoint, parameters: dict, env_parameters: dict = None, session=None, username=None, check_access: bool = True) -> str: @@ -456,6 +482,7 @@ def build_command(entry_point: AppEntryPoint, parameters: dict, # Start with the base command parts = [entry_point.command] + nextflow_run = _is_nextflow_run_command(entry_point.command) # Pass 1: Flagged args in declaration order for p, value in effective: @@ -463,11 +490,30 @@ def build_command(entry_point: AppEntryPoint, parameters: dict, continue validated = _validate_parameter_value(p, value, session=session, username=username, check_access=check_access) + # Nextflow pipeline params (`--foo`) are parsed differently from normal + # CLI switches: a separate leading-dash value becomes another param, and + # omitting a false boolean leaves any pipeline default in effect. Apply + # the Nextflow-safe forms at command-build time too, so cached + # auto-generated manifests from older Fileglancer versions are fixed + # without requiring users to update/re-add the app. + is_nextflow_pipeline_param = nextflow_run and p.flag.startswith("--") + value_separator = "equals" if is_nextflow_pipeline_param else p.value_separator + boolean_style = "value" if is_nextflow_pipeline_param else p.boolean_style if p.type == "boolean": - if value is True: + if boolean_style == "value": + bool_value = "true" if value is True else "false" + if value_separator == "equals": + parts.append(f"{p.flag}={bool_value}") + else: + parts.append(f"{p.flag} {bool_value}") + elif value is True: parts.append(p.flag) else: - parts.append(f"{p.flag} {shlex.quote(validated)}") + quoted = shlex.quote(validated) + if value_separator == "equals": + parts.append(f"{p.flag}={quoted}") + else: + parts.append(f"{p.flag} {quoted}") # Pass 2: Positional args in declaration order for p, value in effective: @@ -488,12 +534,13 @@ def build_command(entry_point: AppEntryPoint, parameters: dict, def collect_path_parameters(entry_point: AppEntryPoint, parameters: dict, - env_parameters: dict = None) -> list[tuple[str, str, str]]: + env_parameters: dict = None) -> list[tuple[AppParameter, str]]: """Collect effective file/directory parameter values needing path validation. Mirrors build_command's effective-value computation (user-provided merged with defaults, across both the env and pipeline namespaces) but returns only - file/directory parameters as (param_key, param_name, raw_value) tuples. + file/directory parameters as (param, raw_value) tuples, so callers can key + validation off the parameter's type and exists flag. Raw (un-expanded) values are returned so that authoritative validation can run in the setuid worker, where '~' and access checks resolve as the target @@ -504,7 +551,7 @@ def collect_path_parameters(entry_point: AppEntryPoint, parameters: dict, param_flat = _flatten_param_items(entry_point.parameters) groups = ((env_flat, env_parameters), (param_flat, parameters)) - result: list[tuple[str, str, str]] = [] + result: list[tuple[AppParameter, str]] = [] for flat, values in groups: for param in flat: if param.type not in ("file", "directory"): @@ -515,5 +562,37 @@ def collect_path_parameters(entry_point: AppEntryPoint, parameters: dict, value = param.default else: continue - result.append((param.key, param.name, str(value))) + result.append((param, str(value))) + return result + + +def collect_creatable_dirs(entry_point: AppEntryPoint, parameters: dict, + env_parameters: dict = None) -> list[tuple[str, str]]: + """Collect effective directory values for params with exists=false. + + Mirrors collect_path_parameters' effective-value logic (user value else + default, across both the env and pipeline namespaces) but returns only + directory params whose path need not exist yet (exists=false), as + (param_name, raw_value) tuples. Raw (un-expanded) values are returned so + the setuid worker resolves '~' as the target user. See submit_job. + """ + env_parameters = env_parameters or {} + env_flat = _flatten_param_items(entry_point.env_parameters) + param_flat = _flatten_param_items(entry_point.parameters) + groups = ((env_flat, env_parameters), (param_flat, parameters)) + + result: list[tuple[str, str]] = [] + for flat, values in groups: + for param in flat: + if param.type != "directory" or param.exists: + continue + if param.key in values: + value = values[param.key] + elif param.default is not None: + value = param.default + else: + continue + if value == "": + continue + result.append((param.name, str(value))) return result diff --git a/fileglancer/apps/jobfiles.py b/fileglancer/apps/jobfiles.py index 33ff0543..b2fb9613 100644 --- a/fileglancer/apps/jobfiles.py +++ b/fileglancer/apps/jobfiles.py @@ -4,6 +4,7 @@ import os import posixpath import re +import shutil from pathlib import Path from typing import Optional @@ -40,6 +41,68 @@ def _resolve_work_dir(db_job: db.JobDB) -> Path: return _build_work_dir(db_job.id, db_job.app_name, db_job.entry_point_id) +def _safe_work_dir_delete_target(db_job: db.JobDB) -> Path: + """Return the normalized work-dir path if it is safe to delete. + + Job work directories are created under ``.fileglancer/jobs`` and include + the job id in their leaf name. Keep deletion constrained to that shape so a + stale or corrupted DB row cannot turn "delete this job" into arbitrary + filesystem removal. + """ + raw_target = Path(os.path.expanduser(os.fspath(db_job.work_dir))) + if not raw_target.is_absolute(): + raise PermissionError( + f"Refusing to delete relative job work directory: {raw_target}" + ) + target = Path(os.path.abspath(os.fspath(raw_target))) + parts = target.parts + under_jobs_root = any( + part == ".fileglancer" + and idx + 2 < len(parts) + and parts[idx + 1] == "jobs" + for idx, part in enumerate(parts) + ) + if not under_jobs_root: + raise PermissionError( + f"Refusing to delete unexpected job work directory: {target}" + ) + + job_id = re.escape(str(db_job.id)) + if not re.search(rf"(^|-){job_id}-", target.name): + raise PermissionError( + f"Refusing to delete job work directory without job id {db_job.id}: {target}" + ) + + return target + + +def delete_job_work_dir(db_job: db.JobDB) -> bool: + """Delete a job's work directory, returning True when something was removed. + + Missing stored paths or already-removed work directories are treated as + already deleted so the DB record can still be cleaned up. Directory symlinks + are unlinked rather than followed. + """ + if not getattr(db_job, "work_dir", None): + return False + + target = _safe_work_dir_delete_target(db_job) + try: + if target.is_symlink(): + target.unlink() + return True + if not target.exists(): + return False + if not target.is_dir(): + raise NotADirectoryError( + f"Job work directory is not a directory: {target}" + ) + shutil.rmtree(target) + return True + except FileNotFoundError: + return False + + def _stored_work_dir_path(db_job: db.JobDB) -> str: """Return the job's work_dir as a stored path string. @@ -86,6 +149,24 @@ def _make_file_info(file_path: str, exists: bool, } +def read_service_url_file(work_dir: Path) -> Optional[str]: + """Read and validate the 'service_url' file in a job work directory.""" + url_file = work_dir / "service_url" + + if not url_file.is_file(): + return None + + try: + url = url_file.read_text().strip() + except OSError: + return None + + if not url.startswith(("http://", "https://")): + return None + + return url + + def get_service_url(db_job: db.JobDB) -> Optional[str]: """Read the service URL from a job's work directory. @@ -98,21 +179,43 @@ def get_service_url(db_job: db.JobDB) -> Optional[str]: if db_job.status != 'RUNNING': return None - work_dir = _resolve_work_dir(db_job) - url_file = work_dir / "service_url" + return read_service_url_file(_resolve_work_dir(db_job)) - if not url_file.is_file(): + +# Startup phases a service job writes to its 'phase' file so the UI can explain +# a wait before the URL appears (chiefly a container image still downloading). +_SERVICE_PHASES = ("pulling_image", "starting") + + +def read_service_phase_file(work_dir: Path) -> Optional[str]: + """Read and validate the 'phase' file in a job work directory.""" + phase_file = work_dir / "phase" + if not phase_file.is_file(): return None try: - url = url_file.read_text().strip() + phase = phase_file.read_text().strip() except OSError: return None - if not url.startswith(("http://", "https://")): + return phase if phase in _SERVICE_PHASES else None + + +def get_service_phase(db_job: db.JobDB) -> Optional[str]: + """Read the current startup phase from a service job's work directory. + + The generated job script writes a short marker (see _SERVICE_PHASES) to a + 'phase' file — e.g. 'pulling_image' while Apptainer downloads the container + image, then 'starting'. This lets the UI show why a service is taking a + while before its URL is published. Returns None unless the job is a RUNNING + service with a recognized phase written. + """ + if getattr(db_job, 'entry_point_type', 'job') != 'service': + return None + if db_job.status != 'RUNNING': return None - return url + return read_service_phase_file(_resolve_work_dir(db_job)) def get_job_file_paths(db_job: db.JobDB) -> dict[str, dict]: @@ -164,6 +267,42 @@ def get_job_file_paths(db_job: db.JobDB) -> dict[str, dict]: return files +# Cap for inline job-file reads. HPC stdout/stderr logs can grow to gigabytes; +# reading one whole would exhaust worker/server memory and blow the 64 MB IPC +# message limit. We return the tail of anything larger (the most useful part of +# a log), with a marker; the file's work-dir browse link gives full access. +_MAX_JOB_FILE_BYTES = 5 * 1024 * 1024 + + +def _read_text_capped(path: Path) -> str: + """Read a text file, capping oversized files to their trailing bytes. + + Files up to _MAX_JOB_FILE_BYTES are returned in full. Larger files return + only their last _MAX_JOB_FILE_BYTES bytes, prefixed with a marker noting how + much was omitted, so a runaway log can't OOM the process or exceed the IPC + limit. Decoding uses errors='replace' so non-UTF-8 log bytes never raise. + """ + size = path.stat().st_size + if size <= _MAX_JOB_FILE_BYTES: + return path.read_text(errors="replace") + with path.open("rb") as f: + f.seek(size - _MAX_JOB_FILE_BYTES) + tail = f.read() + text = tail.decode("utf-8", errors="replace") + # Drop a leading partial line so the tail starts on a clean boundary. + newline = text.find("\n") + if newline != -1: + text = text[newline + 1:] + omitted = size - _MAX_JOB_FILE_BYTES + shown_mb = _MAX_JOB_FILE_BYTES // (1024 * 1024) + header = ( + f"[Fileglancer: this file is {size} bytes; showing only the last " + f"{shown_mb} MB ({omitted} earlier bytes omitted). Open it from the " + f"job's work directory to view the full contents.]\n\n" + ) + return header + text + + def read_job_file(db_job, file_type: str) -> Optional[str]: """Read the content of a job file given a loaded job record. @@ -172,6 +311,7 @@ def read_job_file(db_job, file_type: str) -> Optional[str]: - stdout.log — captured standard output - stderr.log — captured standard error + Oversized files are truncated to their tail (see _read_text_capped). Returns the file content as a string, or None if the file doesn't exist. """ work_dir = _resolve_work_dir(db_job) @@ -182,10 +322,10 @@ def read_job_file(db_job, file_type: str) -> Optional[str]: script_path = getattr(db_job, 'script_path', None) if script_path: path = Path(script_path) - return path.read_text() if path.is_file() else None + return _read_text_capped(path) if path.is_file() else None scripts = sorted(work_dir.glob("*.sh")) if scripts: - return scripts[0].read_text() + return _read_text_capped(scripts[0]) return None elif file_type == "stdout": path = work_dir / "stdout.log" @@ -195,7 +335,7 @@ def read_job_file(db_job, file_type: str) -> Optional[str]: raise ValueError(f"Unknown file type: {file_type}") if path.is_file(): - return path.read_text() + return _read_text_capped(path) return None diff --git a/fileglancer/apps/jobs.py b/fileglancer/apps/jobs.py index 066d91ca..4d84a70f 100644 --- a/fileglancer/apps/jobs.py +++ b/fileglancer/apps/jobs.py @@ -9,7 +9,7 @@ import re import shlex import tempfile -from pathlib import Path +from pathlib import Path, PurePosixPath from datetime import datetime, UTC from typing import Optional @@ -21,18 +21,20 @@ from fileglancer.apps.manifest import ( clone_url_for_stored_app, _dispatch, - _ensure_repo_cache, + ensure_repo_snapshot, get_or_load_manifest, validate_manifest_path, ) from fileglancer.apps.command import ( build_command, build_requirements_check, + collect_creatable_dirs, collect_path_parameters, expand_user_path, merge_requirements, _ENV_VAR_NAME_PATTERN, _URI_PREFIXES, + _WINDOWS_DRIVE_PATTERN, ) from fileglancer.apps.jobfiles import _build_work_dir from fileglancer.giturls import canonical_github_url @@ -116,7 +118,7 @@ async def stop_job_monitor(): def _get_any_active_username(settings) -> str | None: - """Return any username that has active (PENDING/RUNNING) jobs, or None.""" + """Return any username that has non-terminal jobs, or None.""" with db.get_db_session(settings.db_url) as session: active_jobs = db.get_active_jobs(session) for job in active_jobs: @@ -157,7 +159,7 @@ async def _reconnect_as_any_user(settings): continue new_status = info["status"].upper() if new_status != db_job.status: - is_terminal = new_status in ("DONE", "FAILED", "KILLED") + is_terminal = db.is_terminal_job_status(new_status) finished_at = _parse_iso_dt(info.get("finish_time")) if is_terminal else None db.update_job_status( session, db_job.id, new_status, @@ -185,24 +187,49 @@ async def _poll_loop(settings): lock_fd = None try: lock_fd = open(_POLL_LOCK_PATH, "w") - fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) try: - has_jobs = await _poll_jobs(settings) - except Exception: - logger.exception("Error in job poll loop") - has_jobs = True # keep polling on error - # Hold lock through the sleep so no other worker polls - # until this interval is over - await asyncio.sleep(settings.cluster.poll_interval) - fcntl.flock(lock_fd, fcntl.LOCK_UN) - lock_fd.close() + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except OSError: + # Another worker holds the lock this cycle — skip and retry. + lock_fd.close() + lock_fd = None + await asyncio.sleep(settings.cluster.poll_interval) + continue - if not has_jobs: - logger.info("No active jobs — poll loop stopping") - _poll_task = None - return + # try/finally so the lock is always released, even if the task is + # cancelled (e.g. on shutdown) while we hold it. + try: + try: + has_jobs = await _poll_jobs(settings) + except Exception: + logger.exception("Error in job poll loop") + has_jobs = True # keep polling on error + + if not has_jobs: + # No active jobs: stop the loop. Clear _poll_task and return + # with no await in between, so a concurrent submit_job() + # either sees this task still alive or starts a fresh loop — + # there is no gap where _poll_task is set while the loop is + # exiting. Re-check for active jobs immediately before + # returning (again, no await in between) so a job submitted + # during this cycle keeps the loop running rather than being + # left unpolled. + if _get_any_active_username(settings) is None: + logger.info("No active jobs — poll loop stopping") + _poll_task = None + return + # A job appeared mid-cycle; keep polling. + continue + + # Hold the lock through the sleep so a co-worker doesn't + # double-poll within the same interval. + await asyncio.sleep(settings.cluster.poll_interval) + finally: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + lock_fd.close() + lock_fd = None except OSError: - # Another worker is polling this cycle — skip and retry + # Opening the lock file failed — retry next cycle. if lock_fd: lock_fd.close() await asyncio.sleep(settings.cluster.poll_interval) @@ -220,12 +247,15 @@ async def _poll_jobs(settings): if not active_jobs: return False + now_naive = datetime.now(UTC).replace(tzinfo=None) + unknown_timeout_hours = settings.apps.unknown_timeout_hours + # Handle zombie jobs (no cluster_job_id after timeout) jobs_to_poll = [] for db_job in active_jobs: if not db_job.cluster_job_id: created = db_job.created_at.replace(tzinfo=None) if db_job.created_at.tzinfo else db_job.created_at - age_minutes = (datetime.now(UTC).replace(tzinfo=None) - created).total_seconds() / 60 + age_minutes = (now_naive - created).total_seconds() / 60 if age_minutes > settings.cluster.zombie_timeout_minutes: db.update_job_status(session, db_job.id, "FAILED", finished_at=datetime.now(UTC)) logger.warning( @@ -233,6 +263,24 @@ async def _poll_jobs(settings): f"{age_minutes:.0f} minutes, marked FAILED" ) continue + # Give up on jobs stuck in UNKNOWN past the cutoff: the scheduler can + # no longer report them (aged out of the queue/history), so continued + # polling would never resolve them. Measure from status_updated_at + # (when it entered UNKNOWN), falling back to created_at for rows + # predating that column. + if unknown_timeout_hours and db_job.status == "UNKNOWN": + ref = db_job.status_updated_at or db_job.created_at + if ref is not None: + ref_naive = ref.replace(tzinfo=None) if ref.tzinfo else ref + age_hours = (now_naive - ref_naive).total_seconds() / 3600 + if age_hours > unknown_timeout_hours: + db.update_job_status(session, db_job.id, "FAILED", + finished_at=datetime.now(UTC)) + logger.warning( + f"Job {db_job.id} stuck in UNKNOWN for {age_hours:.0f}h " + f"(cutoff {unknown_timeout_hours}h), marked FAILED" + ) + continue jobs_to_poll.append(db_job) if not jobs_to_poll: @@ -278,7 +326,7 @@ async def _poll_jobs(settings): old_status = db_job.status if new_status == old_status: continue - is_terminal = new_status in ("DONE", "FAILED", "KILLED") + is_terminal = db.is_terminal_job_status(new_status) finished_at = _parse_iso_dt(info.get("finish_time")) if is_terminal else None db.update_job_status( session, db_job.id, new_status, @@ -390,6 +438,93 @@ def _container_sif_name(container_url: str) -> str: _DEFAULT_CONTAINER_CACHE_DIR = "$HOME/.fileglancer/apptainer_cache" +def _quote_container_cache_dir(cache_dir: Optional[str], + username: Optional[str] = None) -> str: + """Return a shell-safe APPTAINER_CACHE_DIR assignment value. + + Preferences may use the familiar ``~/...`` spelling shown in the UI. + ``shlex.quote("~/...")`` would make that a literal directory named ``~``, + so expand current-user tildes to the target user's home before quoting. + If the home cannot be resolved (e.g. in a test/dev environment), fall back + to a shell ``$HOME`` prefix so expansion still happens in the user worker. + """ + raw = (cache_dir or "").strip() + if not raw: + return _DEFAULT_CONTAINER_CACHE_DIR + + if raw == "~" or raw.startswith("~/"): + suffix = raw[2:] if raw.startswith("~/") else "" + home = ( + os.path.expanduser(f"~{username}") + if username else os.path.expanduser("~") + ) + if home.startswith("~"): + return "$HOME" if not suffix else f"$HOME/{shlex.quote(suffix)}" + # Join with '/' rather than pathlib: the value lands in a bash script, + # so Windows-style separators must never be introduced here. + expanded = f"{home}/{suffix}" if suffix else home + return shlex.quote(expanded) + + return shlex.quote(os.path.expanduser(raw)) + + +# Runtime helper emitted for service jobs. It allocates a free TCP port on the +# compute node (a port chosen on the submit host would be meaningless there) and +# exports it as FG_SERVICE_PORT along with FG_HOSTNAME, so a service command can +# bind to a known address without reimplementing port discovery. Prefers python +# (authoritative bind-to-0), then falls back to a bash probe of ephemeral ports. +# It also mints FG_SERVICE_TOKEN, a URL-safe random secret the service can use +# for auth (e.g. --token-password="$FG_SERVICE_TOKEN") and that auto_url can +# splice into the published URL via the ${FG_SERVICE_TOKEN} placeholder. +_SERVICE_PORT_HELPER = r"""# Fileglancer service setup: pick a free port, expose the hostname, mint a token +__fg_free_port() { + local p py i + for py in python3 python; do + if command -v "$py" >/dev/null 2>&1; then + p="$("$py" -c 'import socket; s=socket.socket(); s.bind(("",0)); print(s.getsockname()[1]); s.close()' 2>/dev/null)" || true + [ -n "$p" ] && { printf '%s' "$p"; return 0; } + fi + done + for i in $(seq 1 50); do + p=$(( (RANDOM % 16384) + 49152 )) + if ! (exec 3<>"/dev/tcp/127.0.0.1/$p") 2>/dev/null; then + printf '%s' "$p"; return 0 + fi + done + printf '%s' 8080 +} +export FG_HOSTNAME="$(hostname)" +export FG_SERVICE_PORT="$(__fg_free_port)" +export FG_SERVICE_TOKEN="$(openssl rand -hex 24 2>/dev/null || python3 -c 'import secrets; print(secrets.token_hex(24))' 2>/dev/null || date +%s%N | sha256sum | cut -c1-48)" +""" + + +def _build_service_url_publisher(suffix: str = "") -> str: + """Bash that publishes the service URL once $FG_SERVICE_PORT is live. + + Runs in the background so it tolerates a slow start (the container image may + still be pulling); it writes SERVICE_URL_PATH only when the port accepts a + connection, and gives up after a bounded wait, logging to stderr. `suffix` is + appended to http://$FG_HOSTNAME:$FG_SERVICE_PORT and may reference the + ${FG_SERVICE_TOKEN}/${FG_SERVICE_PORT}/${FG_HOSTNAME} shell variables; it is + validated for shell-safety at manifest load (AppEntryPoint), so it is safe to + embed inside the double-quoted printf argument here. + """ + return "\n".join([ + "# Publish the service URL once the port is accepting connections.", + "(", + " for _ in $(seq 1 3600); do", + ' if (exec 3<>"/dev/tcp/127.0.0.1/$FG_SERVICE_PORT") 2>/dev/null; then', + f' printf \'http://%s:%s%s\' "$FG_HOSTNAME" "$FG_SERVICE_PORT" "{suffix}" > "$SERVICE_URL_PATH"', + " exit 0", + " fi", + " sleep 1", + " done", + ' echo "Fileglancer: port $FG_SERVICE_PORT never opened; service URL not published." >&2', + ") &", + ]) + + def _build_container_script( container_url: str, command: str, @@ -397,10 +532,14 @@ def _build_container_script( bind_paths: list[str], container_args: Optional[str] = None, cache_dir: Optional[str] = None, + username: Optional[str] = None, ) -> str: """Build shell script for running a command inside an Apptainer container.""" sif_name = _container_sif_name(container_url) - docker_url = container_url if container_url.startswith("docker://") else f"docker://{container_url}" + docker_url = ( + container_url + if container_url.startswith("docker://") else f"docker://{container_url}" + ) # Deduplicate and sort bind paths all_binds = sorted(set([work_dir] + bind_paths)) @@ -412,7 +551,7 @@ def _build_container_script( split_args = shlex.split(container_args) extra = " " + " ".join(shlex.quote(arg) for arg in split_args) - resolved_dir = shlex.quote(cache_dir) if cache_dir else _DEFAULT_CONTAINER_CACHE_DIR + resolved_dir = _quote_container_cache_dir(cache_dir, username=username) lines = [ "# Apptainer container setup", @@ -420,14 +559,71 @@ def _build_container_script( 'mkdir -p "$APPTAINER_CACHE_DIR"', f'SIF_PATH="$APPTAINER_CACHE_DIR/{sif_name}"', 'if [ ! -f "$SIF_PATH" ]; then', - f' apptainer pull "$SIF_PATH" {shlex.quote(docker_url)}', + # Report the (often multi-minute) image download so the UI can say so. + # FG_PHASE_PATH is set in the preamble; guard in case it is not. + ' [ -n "$FG_PHASE_PATH" ] && printf pulling_image > "$FG_PHASE_PATH" 2>/dev/null || true', + # --disable-cache: the built SIF here is the only copy we keep. We pull + # only when it's missing, so Apptainer's own layer/SIF cache would just + # duplicate gigabytes to speed up a re-pull that rarely happens. Skipping + # it means a re-pull (if this SIF is deleted) re-downloads from the + # registry, which is the right trade for not double-storing every image. + f' apptainer pull --disable-cache "$SIF_PATH" {shlex.quote(docker_url)}', 'fi', + '[ -n "$FG_PHASE_PATH" ] && printf starting > "$FG_PHASE_PATH" 2>/dev/null || true', f'apptainer exec {bind_flags}{extra} "$SIF_PATH" \\', f' {command}', ] return "\n".join(lines) +def _container_bind_paths(entry_point, parameters: dict, + env_parameters: Optional[dict], username: Optional[str], + cached_repo_dir) -> list[str]: + """Compute the host paths to bind-mount into a container runnable. + + Binds are drawn from three sources, in order: + + 1. Each effective file/directory parameter's value (a file binds its parent + dir). "Effective" means the same set the command is built from — user + values merged with manifest defaults, across BOTH the pipeline + (`parameters`) and env-tab (`env_parameters`) namespaces — so a file + default or an env-tab file parameter is mounted rather than left + dangling. Cloud-storage URIs and non-absolute values are skipped — they + are not bind-mountable and would otherwise produce garbage binds. + 2. The runnable's explicit `bind_paths`. + 3. The cached repo clone, but only when the command runs from `repo`. The + `repo` symlink lives inside the (already-bound) work dir yet points at + the clone outside it, so without this bind the symlink dangles in the + container and the `cd` into it fails. Container runnables default to + `work`, so this is only added when the author opts into `repo`. + + The work dir itself is always bound by `_build_container_script`, so it is + not included here. + """ + bind_paths: list[str] = [] + for param, raw_value in collect_path_parameters( + entry_point, parameters, env_parameters): + expanded = expand_user_path(str(raw_value), username) + # Windows drive paths count as absolute so a dev/test server on Windows + # composes the same script; path validation accepts them the same way. + is_absolute = (expanded.startswith("/") + or _WINDOWS_DRIVE_PATTERN.match(expanded)) + if expanded.startswith(_URI_PREFIXES) or not is_absolute: + continue + if param.type == "directory": + bind_paths.append(expanded) + else: + # PurePosixPath: the bind flag lands in a bash script, and the value + # is already '/'-normalized, so the parent must stay POSIX-style + # even when the server process runs on Windows (dev/test). + bind_paths.append(str(PurePosixPath(expanded).parent)) + if entry_point.bind_paths: + bind_paths.extend(entry_point.bind_paths) + if entry_point.effective_working_dir == "repo": + bind_paths.append(str(cached_repo_dir)) + return bind_paths + + async def submit_job( username: str, app_url: str, @@ -491,21 +687,43 @@ async def submit_job( session=session, username=username, check_access=False) # Authoritative per-user path validation: check that file/directory params - # exist and are readable, run in the setuid worker as the target user. This - # is the same fix applied to data links in commit f9858f48 — the server runs - # as a service account that isn't in the user's groups, so a redundant + # exist and are readable, run in the setuid worker as the target user. The + # server runs as a service account that isn't in the user's groups, so a # server-side check would wrongly reject (or, on local FS, wrongly accept) # paths the user can actually access. + # Create any directory params with exists=false first, as the user, so a + # home default like '~/.fileglancer/logs' exists by the time the job runs. + # Containment (within a file share) is enforced in the worker before + # makedirs, so this never writes outside a share. + creatable_dirs = collect_creatable_dirs(entry_point, parameters, env_parameters) + if creatable_dirs: + paths_to_create = {str(i): value for i, (_, value) in enumerate(creatable_dirs)} + creation = await _dispatch(username, "create_dirs", paths=paths_to_create) + errors = (creation or {}).get("errors") or {} + if errors: + idx = min(int(i) for i in errors) + param_name, _ = creatable_dirs[idx] + raise ValueError(f"Parameter '{param_name}': {errors[str(idx)]}") + path_params = collect_path_parameters(entry_point, parameters, env_parameters) if path_params: - paths_to_check = {str(i): value for i, (_, _, value) in enumerate(path_params)} - validation = await _dispatch(username, "validate_paths", paths=paths_to_check) + paths_to_check = {str(i): value for i, (_, value) in enumerate(path_params)} + # exists=false params are outputs the job may create: containment check + # only. Directory params among them were just created above, but file + # params (e.g. a Nextflow output file) never exist pre-launch. + may_be_missing = [str(i) for i, (param, _) in enumerate(path_params) + if not param.exists] + # Expected type per key, so a folder pasted into a file param (or vice + # versa) is rejected here rather than failing at job runtime. + types = {str(i): param.type for i, (param, _) in enumerate(path_params)} + validation = await _dispatch(username, "validate_paths", paths=paths_to_check, + may_be_missing=may_be_missing, types=types) errors = (validation or {}).get("errors") or {} if errors: # Report the first failure, keyed back to its parameter name, to # match the single-message format build_command would have raised. idx = min(int(i) for i in errors) - _, param_name, _ = path_params[idx] + param_name = path_params[idx][0].name raise ValueError(f"Parameter '{param_name}': {errors[str(idx)]}") # Build resource spec (extra_args passed separately, not from manifest) @@ -535,7 +753,7 @@ async def submit_job( "memory": resource_spec.memory, "walltime": resource_spec.walltime, "queue": resource_spec.queue, - "extra_args": " ".join(resource_spec.extra_args) if resource_spec.extra_args else None, + "extra_args": shlex.join(resource_spec.extra_args) if resource_spec.extra_args else None, }.items() if v is not None } @@ -544,6 +762,9 @@ async def submit_job( # Not in the user's library: clone the URL as given (a bare URL resolves the # current default). Overridden below with the pinned URL when installed. app_clone_url = app_url + app_installed = False + pinned_sha = None + pinned_code_sha = None with db.get_db_session(settings.db_url) as session: # Read user's container cache dir preference @@ -556,9 +777,50 @@ async def submit_job( user_app = db.get_user_app(session, username, app_url, manifest_path) app_name = user_app.name if user_app is not None else manifest.name if user_app is not None: + app_installed = True stored_app_url = user_app.url app_clone_url = clone_url_for_stored_app(stored_app_url, user_app.branch) + pinned_sha = user_app.commit_sha + pinned_code_sha = user_app.code_commit_sha + + # Resolve the immutable snapshot the job will run from. A pinned app finds + # its snapshot already on disk (no git work, no network); a legacy unpinned + # row resolves the branch clone's current HEAD and is pinned to it below, + # so it never drifts again. Pulling is never done here; updates are an + # explicit user action via the "Update" app endpoint. + executed_repo_url = None + if manifest.repo_url and canonical_github_url(manifest.repo_url) != stored_app_url: + # Manifest and tool code live in separate repos: the job runs from the + # code repo's snapshot root. + cached_repo_dir, executed_sha = await ensure_repo_snapshot( + manifest.repo_url, sha=pinned_code_sha, username=username) + cd_suffix = "repo" + executed_repo_url = canonical_github_url(manifest.repo_url) + if app_installed and (pinned_sha is None or pinned_code_sha is None): + app_sha = pinned_sha + if app_sha is None: + # Pin the manifest repo as well — left unpinned, this app's + # manifest would keep drifting with the shared branch clone + # and update checks would skip it forever. + _, app_sha = await ensure_repo_snapshot( + app_clone_url, username=username) + with db.get_db_session(settings.db_url) as session: + db.set_user_app_pins(session, username, stored_app_url, + manifest_path, + commit_sha=app_sha, + code_commit_sha=executed_sha) + else: + # Manifest and tool code share one repo: run from the subdirectory + # that contains the manifest. + cached_repo_dir, executed_sha = await ensure_repo_snapshot( + app_clone_url, sha=pinned_sha, username=username) + cd_suffix = f"repo/{manifest_path}" if manifest_path else "repo" + if app_installed and pinned_sha is None: + with db.get_db_session(settings.db_url) as session: + db.set_user_app_pins(session, username, stored_app_url, + manifest_path, commit_sha=executed_sha) + with db.get_db_session(settings.db_url) as session: db_job = db.create_job( session=session, username=username, @@ -579,6 +841,8 @@ async def submit_job( command=entry_point.command, conda_env=entry_point.conda_env, requirements=effective_requirements, + commit_sha=executed_sha, + code_repo_url=executed_repo_url, ) job_id = db_job.id @@ -589,124 +853,114 @@ async def submit_job( db_job.work_dir = str(work_dir) session.commit() - # Ensure the repo is cached in the user's cache (~username/.fileglancer/apps). - # Pulling is never done here; updates are an explicit user action via the - # "Update" app endpoint. The manifest read above already reflects the cache. - if manifest.repo_url and canonical_github_url(manifest.repo_url) != stored_app_url: - # Manifest and tool code live in separate repos: cache the code repo - # and run from its root. - cached_repo_dir = await _ensure_repo_cache(manifest.repo_url, username=username) - cd_suffix = "repo" - else: - # Manifest and tool code share one repo: cache it and run from the - # subdirectory that contains the manifest. - cached_repo_dir = await _ensure_repo_cache(app_clone_url, username=username) - cd_suffix = f"repo/{manifest_path}" if manifest_path else "repo" + # The job row exists but the cluster knows nothing about it yet: any + # failure between here and a successful worker submit (env-var + # validation, script assembly, the submit dispatch) must remove the + # row, or it lingers as a phantom PENDING job that never runs. + try: + # Build environment variable export lines + env_lines = "" + if merged_env: + parts = [] + for var_name, var_value in merged_env.items(): + if not _ENV_VAR_NAME_PATTERN.match(var_name): + raise ValueError(f"Invalid environment variable name: '{var_name}'") + parts.append(f"export {var_name}={shlex.quote(var_value)}") + env_lines = "\n".join(parts) + "\n" + + # Set up the script preamble: + # - FG_WORK_DIR: the job's working directory (used by subsequent variables) + # - Unset PIXI_PROJECT_MANIFEST so pixi uses the repo's own manifest + # - SERVICE_URL_PATH: for service-type jobs, where to write the service URL + # - cd into the repo so commands can find project files (pixi.toml, scripts, etc.) + preamble_lines = [ + "unset PIXI_PROJECT_MANIFEST", + f"export FG_WORK_DIR={shlex.quote(str(work_dir))}", + # Where the script reports its startup phase (e.g. pulling a container + # image). The UI reads this to explain a wait before a service is ready. + 'export FG_PHASE_PATH="$FG_WORK_DIR/phase"', + ] + # For local executor, trap EXIT to write the exit code to a file so + # PID-based polling can determine the final status after the process exits. + if settings.cluster.executor == "local": + preamble_lines.append( + 'trap \'echo $? > "$FG_WORK_DIR/exit_code"\' EXIT' + ) + if settings.apps.extra_paths: + path_suffix = os.pathsep.join(shlex.quote(p) for p in settings.apps.extra_paths) + preamble_lines.append(f"export PATH=$PATH:{path_suffix}") + if entry_point.type == "service": + preamble_lines.append('export SERVICE_URL_PATH="$FG_WORK_DIR/service_url"') + preamble_lines.append(_SERVICE_PORT_HELPER) + # With auto_url, Fileglancer publishes the URL for the author: a + # background probe waits for $FG_SERVICE_PORT to accept connections and + # then writes http://$FG_HOSTNAME:$FG_SERVICE_PORT plus the optional + # (validated) service_url_suffix. A service that manages its own URL + # leaves auto_url unset and writes SERVICE_URL_PATH itself. + if entry_point.auto_url: + preamble_lines.append( + _build_service_url_publisher(entry_point.service_url_suffix or "") + ) + # Choose the working directory. 'work' runs from the job's work dir (the + # repo is still reachable via the `repo` symlink); 'repo' runs from the + # cloned project (optionally the manifest's subdirectory). cd_suffix may + # include a Git-derived directory name, so shell-escape it — FG_WORK_DIR + # stays in its own double-quoted segment so it still expands. + if entry_point.effective_working_dir == "work": + preamble_lines.append('cd "$FG_WORK_DIR"') + else: + preamble_lines.append(f'cd "$FG_WORK_DIR"/{shlex.quote(cd_suffix)}') + script_parts = ["\n".join(preamble_lines)] + + # Conda environment activation + if entry_point.conda_env: + conda_activation = ( + 'eval "$(conda shell.bash hook)"\n' + f'conda activate {shlex.quote(entry_point.conda_env)}' + ) + script_parts.append(conda_activation) - # Build environment variable export lines - env_lines = "" - if merged_env: - parts = [] - for var_name, var_value in merged_env.items(): - if not _ENV_VAR_NAME_PATTERN.match(var_name): - raise ValueError(f"Invalid environment variable name: '{var_name}'") - parts.append(f"export {var_name}={shlex.quote(var_value)}") - env_lines = "\n".join(parts) + "\n" - - # Set up the script preamble: - # - FG_WORK_DIR: the job's working directory (used by subsequent variables) - # - Unset PIXI_PROJECT_MANIFEST so pixi uses the repo's own manifest - # - SERVICE_URL_PATH: for service-type jobs, where to write the service URL - # - cd into the repo so commands can find project files (pixi.toml, scripts, etc.) - preamble_lines = [ - "unset PIXI_PROJECT_MANIFEST", - f"export FG_WORK_DIR={shlex.quote(str(work_dir))}", - ] - # For local executor, trap EXIT to write the exit code to a file so - # PID-based polling can determine the final status after the process exits. - if settings.cluster.executor == "local": - preamble_lines.append( - 'trap \'echo $? > "$FG_WORK_DIR/exit_code"\' EXIT' - ) - if settings.apps.extra_paths: - path_suffix = os.pathsep.join(shlex.quote(p) for p in settings.apps.extra_paths) - preamble_lines.append(f"export PATH=$PATH:{path_suffix}") - if entry_point.type == "service": - preamble_lines.append('export SERVICE_URL_PATH="$FG_WORK_DIR/service_url"') - # Choose the working directory. 'work' runs from the job's work dir (the - # repo is still reachable via the `repo` symlink); 'repo' runs from the - # cloned project (optionally the manifest's subdirectory). cd_suffix may - # include a Git-derived directory name, so shell-escape it — FG_WORK_DIR - # stays in its own double-quoted segment so it still expands. - if entry_point.effective_working_dir == "work": - preamble_lines.append('cd "$FG_WORK_DIR"') - else: - preamble_lines.append(f'cd "$FG_WORK_DIR"/{shlex.quote(cd_suffix)}') - script_parts = ["\n".join(preamble_lines)] - - # Conda environment activation - if entry_point.conda_env: - conda_activation = ( - 'eval "$(conda shell.bash hook)"\n' - f'conda activate {shlex.quote(entry_point.conda_env)}' - ) - script_parts.append(conda_activation) - - # If container is defined, wrap command in apptainer exec - if effective_container: - bind_paths = [] - for param in entry_point.flat_parameters(): - if param.type in ("file", "directory") and param.key in parameters: - path_val = str(parameters[param.key]) - # Expand ~ against the user's home (same normalization as the - # command). Skip cloud-storage URIs and anything that isn't an - # absolute local path — those are not bind-mountable and would - # otherwise produce garbage binds (e.g. s3://bucket/k -> s3:/bucket). - expanded = expand_user_path(path_val, username) - if expanded.startswith(_URI_PREFIXES) or not expanded.startswith("/"): - continue - if param.type == "directory": - bind_paths.append(expanded) - else: - bind_paths.append(str(Path(expanded).parent)) - if entry_point.bind_paths: - bind_paths.extend(entry_point.bind_paths) - - command = _build_container_script( - container_url=effective_container, - command=command, - work_dir=str(work_dir), - bind_paths=bind_paths, - container_args=effective_container_args, - cache_dir=container_cache_dir, - ) + # If container is defined, wrap command in apptainer exec + if effective_container: + bind_paths = _container_bind_paths( + entry_point, parameters, env_parameters, username, cached_repo_dir + ) - if env_lines: - script_parts.append(env_lines.rstrip()) - # Verify required tools now that PATH, conda, and env vars are set up, but - # before pre_run/command do any real work. Fails the job with a readable - # message in stderr if a requirement is unmet. - req_check = build_requirements_check(effective_requirements) - if req_check: - script_parts.append(req_check) - if effective_pre_run: - script_parts.append(effective_pre_run.rstrip()) - script_parts.append(command) - if effective_post_run: - script_parts.append(effective_post_run.rstrip()) - full_command = "\n\n".join(script_parts) - - # Set work_dir and log paths on resource spec - resource_spec.work_dir = str(work_dir) - resource_spec.stdout_path = str(work_dir / "stdout.log") - resource_spec.stderr_path = str(work_dir / "stderr.log") - - # Submit to the cluster as the target user via the persistent worker: - # it creates the work directory, symlinks the repo, and calls - # executor.submit() — all with the user's identity. - job_name = f"{manifest.name}-{entry_point.id}" - cluster_config = settings.cluster.model_dump(exclude_none=True) - try: + command = _build_container_script( + container_url=effective_container, + command=command, + work_dir=str(work_dir), + bind_paths=bind_paths, + container_args=effective_container_args, + cache_dir=container_cache_dir, + username=username, + ) + + if env_lines: + script_parts.append(env_lines.rstrip()) + # Verify required tools now that PATH, conda, and env vars are set up, but + # before pre_run/command do any real work. Fails the job with a readable + # message in stderr if a requirement is unmet. + req_check = build_requirements_check(effective_requirements) + if req_check: + script_parts.append(req_check) + if effective_pre_run: + script_parts.append(effective_pre_run.rstrip()) + script_parts.append(command) + if effective_post_run: + script_parts.append(effective_post_run.rstrip()) + full_command = "\n\n".join(script_parts) + + # Set work_dir and log paths on resource spec + resource_spec.work_dir = str(work_dir) + resource_spec.stdout_path = str(work_dir / "stdout.log") + resource_spec.stderr_path = str(work_dir / "stderr.log") + + # Submit to the cluster as the target user via the persistent worker: + # it creates the work directory, symlinks the repo, and calls + # executor.submit() — all with the user's identity. + job_name = f"{manifest.name}-{entry_point.id}" + cluster_config = settings.cluster.model_dump(exclude_none=True) worker_result = await _dispatch( username, "submit", cluster_config=cluster_config, @@ -728,8 +982,6 @@ async def submit_job( cached_repo_dir=str(cached_repo_dir), ) except Exception: - # Cluster submission failed — remove the PENDING DB record so - # the job does not appear in the user's jobs list. with db.get_db_session(settings.db_url) as session: db.delete_job(session, job_id, username) raise @@ -790,7 +1042,12 @@ def _build_resource_spec(entry_point: AppEntryPoint, overrides: Optional[dict], if overrides.get("queue") is not None: queue = overrides["queue"] if overrides.get("extra_args") is not None: - extra_args = [overrides["extra_args"]] + # The UI/preferences deliver extra_args as one string (e.g. + # '-P proj -R "select[mem>8000]"'); split into individual argv + # tokens so the scheduler receives distinct options rather than a + # single malformed argument. Quotes group tokens that contain + # spaces. + extra_args = shlex.split(overrides["extra_args"]) return ResourceSpec( cpus=cpus, @@ -802,18 +1059,33 @@ def _build_resource_spec(entry_point: AppEntryPoint, overrides: Optional[dict], async def cancel_job(job_id: int, username: str) -> db.JobDB: - """Cancel a running or pending job.""" + """Cancel a non-terminal job.""" settings = get_settings() with db.get_db_session(settings.db_url) as session: db_job = db.get_job(session, job_id, username) if db_job is None: raise ValueError(f"Job {job_id} not found") - if db_job.status not in ("PENDING", "RUNNING"): + if db.is_terminal_job_status(db_job.status): raise ValueError(f"Job {job_id} is not cancellable (status: {db_job.status})") - # Cancel on cluster as the target user - if db_job.cluster_job_id: + # Actually stop the running job as the target user. The local executor + # spawns a bash subprocess (plus its child workload) whose PID a fresh + # executor can't reach, so kill it and its whole process tree by the + # PID persisted in its work dir; other executors (LSF, ...) cancel by + # cluster job id via py-cluster-api. + if settings.cluster.executor == "local": + if db_job.work_dir: + result = await _dispatch( + username, "cancel_local", work_dir=db_job.work_dir) + # Only record KILLED once the workload is confirmed gone — + # otherwise we'd report success while the job keeps running. + if not (result or {}).get("terminated", False): + raise ValueError( + f"Could not confirm job {job_id} was stopped; it may " + f"still be running. Try again." + ) + elif db_job.cluster_job_id: cluster_config = settings.cluster.model_dump(exclude_none=True) await _dispatch( username, "cancel", diff --git a/fileglancer/apps/manifest.py b/fileglancer/apps/manifest.py index 6e4c8307..7a61ccd6 100644 --- a/fileglancer/apps/manifest.py +++ b/fileglancer/apps/manifest.py @@ -2,7 +2,10 @@ import asyncio import os +import re import shutil +import threading +import time from contextlib import suppress from pathlib import Path, PurePosixPath @@ -46,6 +49,21 @@ async def _dispatch(username: str, action: str, **kwargs) -> dict: _MANIFEST_FILENAME = "runnables.yaml" +# Immutable per-commit checkouts live under //.snapshots/. +# The dirname starts with '.' so it can never collide with a branch-named +# clone directory (git forbids ref components that begin with a dot). +_SNAPSHOTS_DIRNAME = ".snapshots" +_SHA_PATTERN = re.compile(r"^[0-9a-f]{40}$") + + +def validate_commit_sha(sha: str) -> str: + """Validate a full 40-hex commit SHA. It becomes a path component and a + git argument, so anything else is rejected.""" + if not sha or not _SHA_PATTERN.fullmatch(sha): + raise ValueError(f"Invalid commit SHA: '{sha}'") + return sha + + def _repo_cache_base(username: str | None = None) -> Path: """Return the repo cache base directory, optionally for a specific user.""" if username: @@ -355,6 +373,331 @@ async def _ensure_repo_cache(url: str, pull: bool = False, return repo_dir +# --- Immutable per-commit snapshots ----------------------------------------- +# +# Apps are pinned to a commit (user_apps.commit_sha) and jobs run from an +# immutable checkout of that commit, not from the mutable branch clone. The +# branch clone remains the fetch target; snapshots are materialized from it as +# local hardlink clones, so they are cheap in time and disk while still being +# fully self-contained (their .git works when bind-mounted into a container). +# Updating an app creates a new snapshot and repoints the row — sibling apps +# in the same repo and already-running jobs keep the tree they were pinned to. + + +def _snapshots_dir(owner: str, repo: str) -> Path: + return _repo_cache_base() / owner / repo / _SNAPSHOTS_DIRNAME + + +async def _git_head_sha(repo_dir: Path) -> str: + stdout, _ = await _run_git(["git", "-C", str(repo_dir), "rev-parse", "HEAD"]) + return stdout.decode().strip() + + +async def _create_snapshot(owner: str, repo: str, repo_dir: Path, sha: str, + snapshot_dir: Path) -> None: + """Materialize an immutable checkout of sha from the branch clone. + + Builds the snapshot in a sibling .tmp dir and renames it into place, so a + half-built snapshot is never observable at the final path. Caller holds + the per-snapshot lock. + """ + # The pinned commit may predate the branch clone's current tip (a sibling + # app's update pulled past it). If the object isn't present locally, fetch + # it explicitly — GitHub serves reachable commits by SHA. + try: + await _run_git(["git", "-C", str(repo_dir), "cat-file", "-e", + f"{sha}^{{commit}}"]) + except ValueError: + try: + await _run_git(["git", "-C", str(repo_dir), "fetch", "origin", sha], + timeout=120, extra_env=_SSH_GIT_ENV) + except ValueError as e: + raise ValueError( + f"Commit {sha[:7]} is no longer available in {owner}/{repo} " + f"(its history may have been rewritten). Update the app to " + f"pin it to the current revision. ({e})" + ) + + snapshot_dir.parent.mkdir(parents=True, exist_ok=True) + tmp_dir = snapshot_dir.parent / f".tmp-{sha}" + shutil.rmtree(tmp_dir, ignore_errors=True) + try: + # Local clone: objects are hardlinked. --no-checkout because the + # branch clone's HEAD may be detached (tag-pinned apps), which some + # git versions refuse to check out during clone; we check out the + # exact commit ourselves either way. + await _run_git(["git", "clone", "--no-checkout", str(repo_dir), + str(tmp_dir)], timeout=300) + await _run_git(["git", "-C", str(tmp_dir), "checkout", "--detach", sha], + timeout=300) + # Keep the commit reachable inside the snapshot regardless of where + # its branches move, so nothing can ever gc it out from under a job. + await _run_git(["git", "-C", str(tmp_dir), "tag", "-f", + "fileglancer-pin", sha]) + # Point origin at GitHub rather than at the local branch clone. + https_url, _ = _github_remote_urls(owner, repo) + await _run_git(["git", "-C", str(tmp_dir), "remote", "set-url", + "origin", https_url]) + try: + os.rename(tmp_dir, snapshot_dir) + except OSError: + # Lost a creation race — the existing snapshot is equivalent. + if not (snapshot_dir / ".git").exists(): + raise + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +async def ensure_repo_snapshot(url: str, sha: str | None = None, + pull: bool = False, + username: str | None = None) -> tuple[Path, str]: + """Return (snapshot_dir, sha) for an immutable checkout of the repo at sha. + + sha=None means "the branch clone's current HEAD" (after pulling when + pull=True): used at add/update time to pin an app, and at launch time to + backfill legacy unpinned rows. + + When username is provided, the work is delegated to a worker subprocess + running as the target user. + """ + if sha is not None: + validate_commit_sha(sha) + + if username: + result = await _dispatch(username, "ensure_snapshot", url=url, sha=sha, + pull=pull) + return Path(result["snapshot_dir"]), result["sha"] + + owner, repo, _ = _parse_github_url(url) + + # Hot path (job launch of a pinned app): the snapshot already exists, so + # no git work — and no network — happens at all. Touch the directory so + # the GC grace period protects a snapshot that is about to be launched + # from but isn't yet referenced by a committed job row. + if sha is not None and not pull: + snapshot_dir = _snapshots_dir(owner, repo) / sha + if (snapshot_dir / ".git").exists(): + with suppress(OSError): + os.utime(snapshot_dir) + return snapshot_dir, sha + + repo_dir = await _ensure_repo_cache(url, pull=pull) + if sha is None: + sha = await _git_head_sha(repo_dir) + snapshot_dir = _snapshots_dir(owner, repo) / sha + + lock = _get_repo_lock(owner, repo, f"snapshot/{sha}") + async with lock: + if (snapshot_dir / ".git").exists(): + with suppress(OSError): + os.utime(snapshot_dir) + return snapshot_dir, sha + # A directory without .git is debris from an interrupted delete. + if snapshot_dir.exists(): + shutil.rmtree(snapshot_dir, ignore_errors=True) + await _create_snapshot(owner, repo, repo_dir, sha, snapshot_dir) + + # Opportunistically resume deleting stale trash left by a gc whose + # background delete was cut short (e.g. worker eviction mid-rmtree). + _delete_dirs_in_background(_stale_trash_dirs(_snapshots_dir(owner, repo))) + return snapshot_dir, sha + + +def _delete_dirs_in_background(paths: list[Path]) -> None: + """Delete directories on a daemon thread. + + Snapshot trees can be large (pixi envs build inside them) and live on NFS; + deleting inline would stall the worker's serial request loop. + """ + if not paths: + return + + def _rm(): + for p in paths: + shutil.rmtree(p, ignore_errors=True) + + threading.Thread(target=_rm, daemon=True, name="fg-snapshot-gc").start() + + +# How recently a snapshot (or .tmp/.trash dir) must have been touched to be +# exempt from gc. Guards the windows the DB keep-set cannot see: a snapshot +# resolved by a launch whose job row isn't committed yet, a .tmp dir another +# process is mid-clone into (the per-repo locks are process-local), and a +# .trash dir another process's delete thread is still working through. +_GC_GRACE_SECONDS = 3600 + + +def _is_recent(path: Path) -> bool: + try: + return (time.time() - path.stat().st_mtime) < _GC_GRACE_SECONDS + except OSError: + # Vanished (e.g. another process's delete finished) — nothing to do. + return True + + +def _stale_trash_dirs(snapshots_dir: Path) -> list[Path]: + """Trash dirs old enough that no delete thread can still be on them.""" + if not snapshots_dir.is_dir(): + return [] + return [ + entry for entry in snapshots_dir.iterdir() + if entry.name.startswith(".trash-") and not _is_recent(entry) + ] + + +async def gc_repo_snapshots(url: str, keep_shas: list[str], + username: str | None = None) -> list[str]: + """Remove snapshots of this repo that no app or retained job references. + + keep_shas is computed by the caller from the database (every user_apps pin + for this owner/repo plus the SHAs of retained jobs). Anything created or + used within the grace period is skipped regardless — the DB can't see a + launch that hasn't committed its job row yet, nor another server process's + in-flight snapshot creation. Directories are renamed out of the way + immediately and deleted in the background. Returns the SHAs removed. + Never touches branch clones. + """ + if username: + result = await _dispatch(username, "gc_snapshots", url=url, + keep_shas=list(keep_shas)) + return result["removed"] + + owner, repo, _ = _parse_github_url(url) + keep = {s for s in keep_shas if s and _SHA_PATTERN.fullmatch(s)} + snapshots_dir = _snapshots_dir(owner, repo) + if not snapshots_dir.is_dir(): + return [] + + removed: list[str] = [] + to_delete: list[Path] = [] + for entry in list(snapshots_dir.iterdir()): + name = entry.name + if name.startswith(".trash-"): + # Leftover from a previous gc whose delete never finished. Recent + # trash may still have an active delete thread — leave it alone. + if not _is_recent(entry): + to_delete.append(entry) + continue + is_tmp = name.startswith(".tmp-") + sha = name[len(".tmp-"):] if is_tmp else name + if not _SHA_PATTERN.fullmatch(sha) or sha in keep: + continue + if _is_recent(entry): + continue + # The per-snapshot lock serializes against an in-flight creation in + # this process: if one was underway, it completes first (and a .tmp + # entry will have been renamed to its final path, making it vanish + # from under us). + async with _get_repo_lock(owner, repo, f"snapshot/{sha}"): + if not entry.exists() or _is_recent(entry): + continue + trash = snapshots_dir / f".trash-{sha}-{os.getpid()}-{len(to_delete)}" + try: + os.rename(entry, trash) + except OSError: + continue + to_delete.append(trash) + if not is_tmp: + removed.append(sha) + + _delete_dirs_in_background(to_delete) + if removed: + logger.info(f"GC removed {len(removed)} snapshot(s) of {owner}/{repo}") + return removed + + +# Timeout per ls-remote for update checks. These run while the user's serial +# worker is occupied (blocking their file browsing etc.), so fail fast — a +# missed badge beats a hung UI when GitHub is slow. +_LS_REMOTE_TIMEOUT = 10 + + +async def get_remote_head(url: str, username: str | None = None) -> str | None: + """Resolve the remote tip commit of the revision baked into a stored URL. + + Used by the update-available check: compares against the app's pinned + commit_sha. Returns None when the remote can't be reached or the revision + no longer exists (no badge rather than an error). A revision that is + itself a commit SHA resolves to itself — such an app can never drift. + + A URL without a revision only reaches here for legacy branch=None rows + (clone_url_for_stored_app makes pinned revisions explicit) and for code + repos referenced by bare manifest repo_urls; both track the remote's + *current default*, so resolve HEAD rather than assuming "main". + """ + if username: + result = await _dispatch(username, "remote_heads", urls=[url]) + return (result.get("shas") or {}).get(url) + + owner, repo, branch = _parse_github_url(url) + if branch and _SHA_PATTERN.fullmatch(branch): + return branch + + if branch: + patterns = [f"refs/heads/{branch}", f"refs/tags/{branch}"] + # Branch tips win; for annotated tags prefer the peeled commit + # (refs/tags/x^{}) that ls-remote emits alongside the tag object. + preference = (f"refs/heads/{branch}", + f"refs/tags/{branch}^{{}}", + f"refs/tags/{branch}") + else: + patterns = ["HEAD"] + preference = ("HEAD",) + + https_url, ssh_url = _github_remote_urls(owner, repo) + for remote, extra_env in ((https_url, None), (ssh_url, _SSH_GIT_ENV)): + try: + stdout, _ = await _run_git( + ["git", "ls-remote", remote, *patterns], + timeout=_LS_REMOTE_TIMEOUT, extra_env=extra_env, + ) + except ValueError as e: + # Private repo over HTTPS: retry via SSH; anything else, give up. + if _is_git_auth_error(str(e)): + continue + return None + except Exception: + return None + + shas_by_ref: dict[str, str] = {} + for line in stdout.decode().splitlines(): + parts = line.split() + if len(parts) == 2: + shas_by_ref[parts[1]] = parts[0] + for ref in preference: + sha = shas_by_ref.get(ref) + if sha and _SHA_PATTERN.fullmatch(sha): + return sha + return None + return None + + +async def get_remote_heads(urls: list[str], + username: str | None = None) -> dict[str, str | None]: + """Resolve remote tips for several stored URLs in one worker round-trip. + + The per-user worker handles one request at a time, so N sequential + dispatches would occupy it for N ls-remote timeouts; batching bounds the + occupancy to the slowest single lookup (they run concurrently in-process). + """ + urls = list(dict.fromkeys(urls)) + if not urls: + return {} + + if username: + result = await _dispatch(username, "remote_heads", urls=urls) + shas = result.get("shas") or {} + return {url: shas.get(url) for url in urls} + + results = await asyncio.gather( + *(get_remote_head(url) for url in urls), return_exceptions=True, + ) + return { + url: (sha if isinstance(sha, str) else None) + for url, sha in zip(urls, results) + } + + _SKIP_DIRS = {'.git', 'node_modules', '__pycache__', '.pixi', '.venv', 'venv'} @@ -449,14 +792,15 @@ def _find_manifests_in_repo(repo_dir: Path) -> list[tuple[str, AppManifest]]: async def discover_app_manifests( url: str, username: str | None = None, -) -> tuple[str, list[tuple[str, AppManifest]]]: +) -> tuple[str, str | None, list[tuple[str, AppManifest]]]: """Clone/pull a GitHub repo and discover all manifest files. - Returns (resolved_branch, [(relative_dir_path, AppManifest), ...]). The - resolved branch is the revision actually cloned — resolved in the same + Returns (resolved_branch, head_sha, [(relative_dir_path, AppManifest), ...]). + The resolved branch is the revision actually cloned — resolved in the same process that does the clone, so a private repo's real default branch is used - rather than a fallback. Raises ValueError if the URL is invalid or the clone - fails. + rather than a fallback. head_sha is the commit at the tip of that revision, + used to pin apps added from this discovery. Raises ValueError if the URL is + invalid or the clone fails. When username is provided, the work is delegated to a worker subprocess running as the target user (which holds the user's SSH credentials). @@ -467,37 +811,69 @@ async def discover_app_manifests( (item["path"], AppManifest(**item["manifest"])) for item in result["manifests"] ] - return result["branch"], manifests + return result["branch"], result.get("head_sha"), manifests repo_dir = await _ensure_repo_cache(url, pull=True) owner, repo, _ = _parse_github_url(url) branch = repo_dir.relative_to( (_repo_cache_base() / owner / repo).resolve() ).as_posix() - return branch, _find_manifests_in_repo(repo_dir) + head_sha = await _git_head_sha(repo_dir) + return branch, head_sha, _find_manifests_in_repo(repo_dir) async def fetch_app_manifest(url: str, manifest_path: str = "", - username: str | None = None) -> AppManifest: + username: str | None = None, + sha: str | None = None) -> AppManifest: """Fetch and validate an app manifest from a cloned repo. - Clones the repo if needed, then reads the manifest from disk. + With sha, the manifest is read from the immutable snapshot of that commit + (materializing it if needed) — a pinned app's manifest never drifts just + because the branch clone was pulled past its pin. Without sha, reads from + the mutable branch clone (preview semantics), cloning if needed. When username is provided, the work is delegated to a worker subprocess running as the target user. """ # Reject traversal/unsafe input early, before any worker round-trip. validate_manifest_path(manifest_path) + if sha is not None: + validate_commit_sha(sha) if username: - result = await _dispatch(username, "read_manifest", url=url, manifest_path=manifest_path) + result = await _dispatch(username, "read_manifest", url=url, + manifest_path=manifest_path, sha=sha) return AppManifest(**result["manifest"]) - repo_dir = await _ensure_repo_cache(url) + if sha is not None: + repo_dir, _ = await ensure_repo_snapshot(url, sha=sha) + else: + repo_dir = await _ensure_repo_cache(url) target_dir = _safe_repo_subdir(repo_dir, manifest_path) return _read_manifest_file(target_dir) +async def _fetch_manifest_with_pin_fallback(fetch_url: str, manifest_path: str, + username: str | None, + sha: str | None) -> AppManifest: + """Read the manifest at the pinned commit, falling back to the branch + clone when the pin can't be materialized (snapshot lost and the commit + rewritten away upstream). A drifted manifest beats an unusable app; the + next successful update re-pins. + """ + if sha is None: + return await fetch_app_manifest(fetch_url, manifest_path, username=username) + try: + return await fetch_app_manifest(fetch_url, manifest_path, + username=username, sha=sha) + except Exception as e: + logger.warning( + f"Pinned manifest read failed for {fetch_url}@{sha[:7]} ({e}); " + f"falling back to the branch clone" + ) + return await fetch_app_manifest(fetch_url, manifest_path, username=username) + + async def get_or_load_manifest(username: str, url: str, manifest_path: str = "") -> AppManifest: """Return the manifest for an app, preferring the DB cache. @@ -523,6 +899,7 @@ async def get_or_load_manifest(username: str, url: str, row_exists = row is not None row_url = row.url if row is not None else url row_branch = row.branch if row is not None else None + row_sha = row.commit_sha if row is not None else None if stored is not None: try: @@ -531,18 +908,18 @@ async def get_or_load_manifest(username: str, url: str, logger.warning(f"Stored manifest schema mismatch for {url}: {e}") fetch_url = clone_url_for_stored_app(row_url, row_branch) if row_exists else url - manifest = await fetch_app_manifest(fetch_url, manifest_path, username=username) + manifest = await _fetch_manifest_with_pin_fallback( + fetch_url, manifest_path, username, row_sha) if row_exists: - # branch=None: this is a cache refresh, so leave the requested revision - # (the branch column) untouched. + # Cache refresh: sync only the manifest column. Name/description stay + # untouched — they may be user-chosen (e.g. a custom catalog name) — + # as do the requested revision and pins. with db.get_db_session(settings.db_url) as session: - db.upsert_user_app( + db.update_user_app_manifest_cache( session, username, url=row_url, manifest_path=manifest_path, - name=manifest.name, description=manifest.description, manifest=manifest.model_dump(mode="json"), - bump_updated_at=False, ) return manifest @@ -569,16 +946,20 @@ async def refresh_cached_manifest(username: str, url: str, row_exists = row is not None row_url = row.url if row_exists else url row_branch = row.branch if row_exists else None + row_sha = row.commit_sha if row_exists else None fetch_url = clone_url_for_stored_app(row_url, row_branch) if row_exists else url - manifest = await fetch_app_manifest(fetch_url, manifest_path, username=username) + manifest = await _fetch_manifest_with_pin_fallback( + fetch_url, manifest_path, username, row_sha) with db.get_db_session(settings.db_url) as session: if row_exists: - db.upsert_user_app( + # Sync only the manifest column: name/description may be + # user-chosen (e.g. a custom catalog name) and must survive + # cache refreshes. + db.update_user_app_manifest_cache( session, username, url=row_url, manifest_path=manifest_path, - name=manifest.name, description=manifest.description, manifest=manifest.model_dump(mode="json"), bump_updated_at=bump_updated_at, ) diff --git a/fileglancer/apps/nextflow.py b/fileglancer/apps/nextflow.py index d64b5f03..4f9d234e 100644 --- a/fileglancer/apps/nextflow.py +++ b/fileglancer/apps/nextflow.py @@ -44,6 +44,12 @@ def _convert_property(name: str, prop: dict, is_required: bool) -> AppParameter: "flag": f"--{name}", "name": name.replace("_", " ").title(), "type": param_type, + # Nextflow treats a separate value that starts with '-' as another + # Nextflow/pipeline option (`--runtime_opts --nv` becomes + # params.runtime_opts=true and params.nv=true). Join generated + # pipeline params with '=' so leading-dash values stay attached to the + # intended parameter (`--runtime_opts=--nv`). + "value_separator": "equals", } desc = prop.get("description") @@ -53,6 +59,18 @@ def _convert_property(name: str, prop: dict, is_required: bool) -> AppParameter: if is_required: kwargs["required"] = True + # Nextflow boolean params are not generic CLI switches: omitting a false + # value leaves any pipeline default in effect (e.g. a default true stays + # true). Emit explicit true/false values for schema-derived booleans. + if param_type == "boolean": + kwargs["boolean_style"] = "value" + + # Nextflow schemas only require a path to exist when "exists": true is set; + # anything else (outdir, report paths, ...) is an output the pipeline + # creates, so don't demand it exists before launch. + if param_type in ("file", "directory"): + kwargs["exists"] = prop.get("exists") is True + if "default" in prop: default = prop["default"] if isinstance(default, str): @@ -99,15 +117,16 @@ def convert(self, directory: Path) -> AppManifest: schema_path = directory / _NEXTFLOW_SCHEMA_FILENAME schema = json.loads(schema_path.read_text()) - # Determine app metadata — use owner/repo from the cache path - # (directory is {cache_base}/{owner}/{repo}/{branch}) + # Determine app metadata — use the repo name from the cache path + # (directory is {cache_base}/{owner}/{repo}/{branch}, where branch may + # span multiple path segments) try: from fileglancer.apps.manifest import _repo_cache_base cache_base = _repo_cache_base().resolve() relative = directory.resolve().relative_to(cache_base) - name = f"{relative.parts[0]}/{relative.parts[1]}" + name = relative.parts[1] except Exception: - name = f"{directory.parent.parent.name}/{directory.parent.name}" + name = directory.parent.name description = schema.get("description") # Build parameters from definitions, ordered by allOf diff --git a/fileglancer/apps/pixi.py b/fileglancer/apps/pixi.py index 14246ff9..6d5ecb3a 100644 --- a/fileglancer/apps/pixi.py +++ b/fileglancer/apps/pixi.py @@ -153,8 +153,6 @@ def _task_to_entry_point(name: str, task: dict) -> AppEntryPoint | None: # Surface task env vars as entry-point env defaults. Pixi already applies # task env natively when running the task; exporting them in the generated # job script keeps the values visible and overridable via the job's env. - # (Previously these were emitted as `--env:VAR` CLI flags, which `pixi run` - # does not understand and which broke the generated command.) task_env = task.get("env") env = {k: str(v) for k, v in task_env.items()} if task_env else None diff --git a/fileglancer/database.py b/fileglancer/database.py index 4d6bb9c3..5bc72b04 100644 --- a/fileglancer/database.py +++ b/fileglancer/database.py @@ -180,7 +180,18 @@ class JobDB(Base): # detail endpoint build browse links without realpath'ing mounts per read. work_dir_fsp_name = Column(String, nullable=True) work_dir_subpath = Column(String, nullable=True) + # Commit whose code this job executed (the code repo's SHA when the + # manifest declares a separate repo_url, else the app repo's SHA). NULL for + # jobs submitted before commit pinning existed. + commit_sha = Column(String, nullable=True) + # Repo the executed commit belongs to, when it differs from app_url + # (manifests with a separate repo_url). NULL means commit_sha is app_url's. + code_repo_url = Column(String, nullable=True) created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC)) + # When the status column last changed value. Lets the poll loop measure how + # long a job has sat in a non-progressing state (e.g. UNKNOWN) without + # conflating it with created_at. NULL for rows created before this column. + status_updated_at = Column(DateTime, nullable=True) started_at = Column(DateTime, nullable=True) finished_at = Column(DateTime, nullable=True) @@ -196,6 +207,12 @@ class UserAppDB(Base): name = Column(String, nullable=False) description = Column(String, nullable=True) branch = Column(String, nullable=True) + # Commit the app is pinned to: jobs run from an immutable snapshot of this + # SHA, and only an explicit Update moves it. NULL for legacy rows, which + # get pinned on their next launch or update. + commit_sha = Column(String, nullable=True) + # Pin for the manifest's separate code repo (repo_url), when declared. + code_commit_sha = Column(String, nullable=True) manifest = Column(JSON, nullable=True) added_at = Column(DateTime, nullable=False, default=lambda: datetime.now(UTC)) updated_at = Column(DateTime, nullable=True) @@ -945,6 +962,19 @@ def delete_expired_sessions(session: Session): # --- Job database functions --- +TERMINAL_JOB_STATUSES = ("DONE", "FAILED", "KILLED") + + +def is_terminal_job_status(status: str | None) -> bool: + """Return True when a job status means no scheduler work should remain. + + Any status outside this explicit terminal set (including UNKNOWN or a + scheduler-specific transient state) is treated as active so polling, + cancellation, and delete guards do not drop a potentially live job. + """ + return status in TERMINAL_JOB_STATUSES + + def create_job(session: Session, username: str, app_url: str, app_name: str, entry_point_id: str, entry_point_name: str, parameters: Dict, env_parameters: Optional[Dict] = None, @@ -956,7 +986,9 @@ def create_job(session: Session, username: str, app_url: str, app_name: str, container_args: Optional[str] = None, command: Optional[str] = None, conda_env: Optional[str] = None, - requirements: Optional[List[str]] = None) -> JobDB: + requirements: Optional[List[str]] = None, + commit_sha: Optional[str] = None, + code_repo_url: Optional[str] = None) -> JobDB: """Create a new job record""" now = datetime.now(UTC) job = JobDB( @@ -978,8 +1010,11 @@ def create_job(session: Session, username: str, app_url: str, app_name: str, command=command, conda_env=conda_env, requirements=requirements, + commit_sha=commit_sha, + code_repo_url=code_repo_url, status="PENDING", - created_at=now + created_at=now, + status_updated_at=now, ) session.add(job) session.commit() @@ -1000,9 +1035,14 @@ def get_job(session: Session, job_id: int, username: str) -> Optional[JobDB]: def get_active_jobs(session: Session) -> List[JobDB]: - """Get all jobs with PENDING or RUNNING status""" + """Get all jobs that are not known-terminal. + + UNKNOWN and future scheduler-specific statuses are considered active: + until a job reaches DONE/FAILED/KILLED, Fileglancer should keep polling and + must not allow the record to be deleted as if the cluster job were gone. + """ return session.query(JobDB).filter( - JobDB.status.in_(["PENDING", "RUNNING"]) + ~JobDB.status.in_(TERMINAL_JOB_STATUSES) ).all() @@ -1023,6 +1063,11 @@ def update_job_status(session: Session, job_id: int, status: str, job = session.query(JobDB).filter_by(id=job_id).first() if not job: return None + # Record when the status actually changes so the poll loop can tell how long + # a job has been in a stuck state (e.g. UNKNOWN). A no-op update (same + # status) leaves the timestamp untouched, so it marks entry into the state. + if job.status != status: + job.status_updated_at = datetime.now(UTC) job.status = status if exit_code is not None: job.exit_code = exit_code @@ -1053,7 +1098,7 @@ def delete_old_jobs(session: Session, days: int = 30) -> int: """Delete completed/failed jobs older than the specified number of days""" cutoff = datetime.now(UTC) - timedelta(days=days) deleted = session.query(JobDB).filter( - JobDB.status.in_(["DONE", "FAILED", "KILLED"]), + JobDB.status.in_(TERMINAL_JOB_STATUSES), JobDB.created_at < cutoff ).delete(synchronize_session='fetch') session.commit() @@ -1087,6 +1132,8 @@ def upsert_user_app(session: Session, username: str, url: str, name: str, description: Optional[str] = None, branch: Optional[str] = None, + commit_sha: Optional[str] = None, + code_commit_sha: Optional[str] = None, manifest: Optional[Dict] = None, bump_updated_at: bool = True) -> UserAppDB: """Insert or update a user app row. @@ -1102,6 +1149,10 @@ def upsert_user_app(session: Session, username: str, url: str, Pass branch=None to leave an existing row's branch untouched — manifest-cache refreshes use this so they don't clobber the requested revision with a resolved one. + + commit_sha / code_commit_sha are the app's pins (see UserAppDB). Like + branch, None means "leave the existing value untouched" — only add, update + and launch-time backfill move a pin. """ now = datetime.now(UTC) url = canonical_github_url(url) @@ -1114,6 +1165,8 @@ def upsert_user_app(session: Session, username: str, url: str, name=name, description=description, branch=branch, + commit_sha=commit_sha, + code_commit_sha=code_commit_sha, manifest=manifest, added_at=now, ) @@ -1123,6 +1176,10 @@ def upsert_user_app(session: Session, username: str, url: str, row.description = description if branch is not None: row.branch = branch + if commit_sha is not None: + row.commit_sha = commit_sha + if code_commit_sha is not None: + row.code_commit_sha = code_commit_sha row.manifest = manifest if bump_updated_at: row.updated_at = now @@ -1130,6 +1187,48 @@ def upsert_user_app(session: Session, username: str, url: str, return row +def update_user_app_manifest_cache(session: Session, username: str, url: str, + manifest_path: str = "", *, + manifest: Dict, + bump_updated_at: bool = False + ) -> Optional[UserAppDB]: + """Sync only the cached manifest column on an existing row. + + Cache refreshes must not touch user-facing metadata: a catalog app added + under a custom name would otherwise revert to the raw manifest name + whenever its cache is refilled (e.g. after schema drift invalidates the + stored copy). No-op returning None when the row doesn't exist. + """ + row = get_user_app(session, username, url, manifest_path) + if row is None: + return None + row.manifest = manifest + if bump_updated_at: + row.updated_at = datetime.now(UTC) + session.commit() + return row + + +def set_user_app_pins(session: Session, username: str, url: str, + manifest_path: str = "", *, + commit_sha: Optional[str] = None, + code_commit_sha: Optional[str] = None) -> Optional[UserAppDB]: + """Set an app row's commit pins without touching any other field. + + Used by launch-time backfill of legacy unpinned rows. None leaves a pin + unchanged. Returns the row, or None if it doesn't exist. + """ + row = get_user_app(session, username, url, manifest_path) + if row is None: + return None + if commit_sha is not None: + row.commit_sha = commit_sha + if code_commit_sha is not None: + row.code_commit_sha = code_commit_sha + session.commit() + return row + + def delete_user_app(session: Session, username: str, url: str, manifest_path: str = "") -> bool: """Delete a user app row. Returns True if a row was deleted.""" @@ -1198,15 +1297,35 @@ def create_app_listing(session: Session, owner_username: str, url: str, def update_app_listing(session: Session, listing_id: int, owner_username: str, *, name: Optional[str] = None, - description: Optional[str] = None) -> Optional[AppListingDB]: + description: Optional[str] = None, + url: Optional[str] = None, + branch: Optional[str] = None) -> Optional[AppListingDB]: """Update an existing listing's editable metadata. Returns the listing, or - None if it doesn't exist or isn't owned by owner_username.""" + None if it doesn't exist or isn't owned by owner_username. + + url repoints the listing (the caller has already validated that the new + repo/revision still contains the listing's manifest path); branch is the + requested revision that goes with it and is only applied alongside url. + Raises ValueError if the new url collides with another listing by the + same owner (unique on owner/url/manifest_path).""" listing = session.query(AppListingDB).filter_by( id=listing_id, owner_username=owner_username, ).first() if listing is None: return None + if url is not None: + url = canonical_github_url(url) + duplicate = session.query(AppListingDB).filter( + AppListingDB.owner_username == owner_username, + AppListingDB.url == url, + AppListingDB.manifest_path == listing.manifest_path, + AppListingDB.id != listing_id, + ).first() + if duplicate is not None: + raise ValueError("You already have another listing for this app") + listing.url = url + listing.branch = branch if name is not None: listing.name = name if description is not None: diff --git a/fileglancer/model.py b/fileglancer/model.py index 5743ddc8..dc51dcc0 100644 --- a/fileglancer/model.py +++ b/fileglancer/model.py @@ -1,9 +1,12 @@ import re +import shlex from datetime import datetime from typing import Annotated, Any, List, Literal, Optional, Dict, Union from pydantic import BaseModel, Discriminator, Field, HttpUrl, Tag, field_validator, model_validator +from fileglancer.giturls import _parse_github_url + class FileSharePath(BaseModel): """A file share path from the database""" @@ -317,6 +320,12 @@ class NeuroglancerShortLinkResponse(BaseModel): # --- App Manifest Models --- +# Conservative CLI-flag shape: one or two leading dashes, then an alphanumeric +# followed by word characters, dots, or dashes. Flags are emitted into the +# generated job script unquoted, so anything shell-significant is rejected. +_FLAG_PATTERN = re.compile(r'^-{1,2}[A-Za-z0-9][A-Za-z0-9_.-]*$') + + class AppParameter(BaseModel): """A parameter definition for an app entry point""" flag: Optional[str] = Field( @@ -340,6 +349,33 @@ class AppParameter(BaseModel): pattern: Optional[str] = Field(description="Regex validation pattern for string types", default=None) hidden: bool = Field(description="Whether the parameter is hidden by default in the UI", default=False) raw: bool = Field(description="If true, value is appended to the command without shell quoting", default=False) + value_separator: Literal["space", "equals"] = Field( + description=( + "How flagged parameters are joined to their value in the generated " + "command: 'space' emits '--flag value', while 'equals' emits " + "'--flag=value'." + ), + default="space", + ) + boolean_style: Literal["flag", "value"] = Field( + description=( + "How flagged boolean parameters are emitted: 'flag' emits '--flag' " + "for true and omits false, while 'value' emits '--flag true' or " + "'--flag false' (or '--flag=true/false' when value_separator is " + "'equals')." + ), + default="flag", + ) + exists: bool = Field( + description="file/directory params only: when true (the default), the path " + "must exist and be readable before launch. Set false for outputs " + "the job creates — the pre-launch existence check is skipped " + "(file-share containment is still enforced), and directory params " + "are created (as the user, within an allowed file share) before " + "launch, so a home default like '~/.fileglancer/logs' works on " + "first launch", + default=True, + ) @field_validator("flag") @classmethod @@ -350,6 +386,15 @@ def validate_flag(cls, v): stripped = v.lstrip("-") if not stripped: raise ValueError("Flag must have content after dashes") + # Flags are appended to the generated shell command unquoted, and + # the Nextflow adapter derives them from schema property names, so + # constrain them to a conservative CLI-flag shape rather than + # allowing shell-significant characters through. + if not _FLAG_PATTERN.fullmatch(v): + raise ValueError( + f"Flag must look like '-n' or '--long-name' " + f"(letters, digits, '_', '.', '-'), got '{v}'" + ) return v @field_validator("options", mode="before") @@ -363,6 +408,23 @@ def stringify_options(cls, v): return v return [str(item) for item in v] + @model_validator(mode='after') + def validate_exists(self): + # Check the value, not model_fields_set: manifests round-trip through + # model_dump (worker -> server, DB cache), which serializes the True + # default onto every param, so presence of the field is meaningless. + if not self.exists and self.type not in ("file", "directory"): + raise ValueError( + f"exists is only valid on file and directory parameters, " + f"but '{self.name}' has type '{self.type}'" + ) + if self.boolean_style != "flag" and self.type != "boolean": + raise ValueError( + f"boolean_style is only valid on boolean parameters, " + f"but '{self.name}' has type '{self.type}'" + ) + return self + class AppParameterSection(BaseModel): """A collapsible section that groups parameters in the UI""" @@ -436,6 +498,27 @@ class AppEntryPoint(BaseModel): ), default=None, ) + auto_url: bool = Field( + description=( + "For service entry points only: have Fileglancer publish the service " + "URL for you. Once the service's port ($FG_SERVICE_PORT) is accepting " + "connections, Fileglancer writes http://$FG_HOSTNAME:$FG_SERVICE_PORT " + "(plus service_url_suffix, if set) to SERVICE_URL_PATH. Set this when " + "your service binds to the Fileglancer-provided $FG_SERVICE_PORT." + ), + default=False, + ) + service_url_suffix: Optional[str] = Field( + description=( + "For auto_url service entry points: text appended to " + "http://$FG_HOSTNAME:$FG_SERVICE_PORT when publishing the URL, e.g. a " + "path and/or query for one-click auth. May contain the placeholders " + "${FG_SERVICE_TOKEN}, ${FG_SERVICE_PORT}, ${FG_HOSTNAME} (braces " + "required) and literal URL text; nothing else. Example: " + "'/?access_token=${FG_SERVICE_TOKEN}'." + ), + default=None, + ) requirements: List[str] = Field( description="Required tools for this entry point, e.g. ['apptainer']. Merged with manifest-level requirements.", default=[], @@ -474,6 +557,23 @@ def validate_container(cls, v): raise ValueError(f"container URL contains forbidden characters: {v!r}") return v + @field_validator("service_url_suffix") + @classmethod + def validate_service_url_suffix(cls, v): + if v is None: + return v + # Strip the recognized ${FG_*} placeholders, then reject anything that + # would be unsafe or unexpanded inside the double-quoted shell string the + # suffix is emitted into (a stray $, quote, backtick, backslash, newline). + residual = _SERVICE_URL_PLACEHOLDER.sub("", v) + if _SERVICE_URL_UNSAFE.search(residual): + raise ValueError( + "service_url_suffix may contain only literal URL text and the " + "placeholders ${FG_SERVICE_TOKEN}, ${FG_SERVICE_PORT}, " + f"${{FG_HOSTNAME}} (braces required); got: {v!r}" + ) + return v + @field_validator("bind_paths") @classmethod def validate_bind_paths(cls, v): @@ -550,6 +650,10 @@ def check_conda_container_exclusive(self): raise ValueError("conda_env and container are mutually exclusive — use one or the other") if self.bind_paths and not self.container: raise ValueError("bind_paths requires container to be set") + if self.auto_url and self.type != "service": + raise ValueError("auto_url is only valid for service entry points (type: service)") + if self.service_url_suffix is not None and not self.auto_url: + raise ValueError("service_url_suffix requires auto_url to be set") return self @@ -563,6 +667,13 @@ def check_conda_container_exclusive(self): ) _SHELL_METACHAR_PATTERN = re.compile(r'[;&|`$(){}!<>\n\r]') +# Placeholders Fileglancer substitutes into service_url_suffix at runtime. +_SERVICE_URL_PLACEHOLDER = re.compile(r'\$\{(?:FG_SERVICE_TOKEN|FG_SERVICE_PORT|FG_HOSTNAME)\}') +# The suffix is emitted inside a double-quoted shell string, so only these are +# dangerous once the recognized placeholders are removed: a stray $, a double +# quote, a backtick, a backslash, or a newline. Everything else (?, &, =, /, %, +# ...) is literal inside double quotes and valid in a URL. +_SERVICE_URL_UNSAFE = re.compile(r'[$"`\\\n\r]') _CONDA_ENV_NAME_PATTERN = re.compile(r'^[a-zA-Z0-9_.-]+$') _CONDA_ENV_PATH_FORBIDDEN = re.compile(r'[;&|`$(){}!<>\n\r]') @@ -605,19 +716,43 @@ class AppManifest(BaseModel): description="Required tools, e.g. ['pixi>=0.40', 'npm']", default=[], ) - runnables: List[AppEntryPoint] = Field(description="Available entry points for this app") + runnables: List[AppEntryPoint] = Field( + description="Available entry points for this app", min_length=1) @field_validator("requirements") @classmethod def validate_requirements(cls, v): return _validate_requirements(v) + @field_validator("repo_url") + @classmethod + def validate_repo_url(cls, v): + # A separate code repo must be a GitHub URL the same way app URLs are — + # reject anything else at manifest load so authors get a clear error + # instead of a cryptic failure later at launch/update when + # ensure_repo_snapshot tries to parse it. + if v is None: + return v + try: + _parse_github_url(v) + except ValueError as e: + raise ValueError(f"repo_url must be a GitHub repository URL: {e}") + return v + class UserApp(BaseModel): """A user's saved app reference""" url: str = Field(description="URL to the app manifest") manifest_path: str = Field(description="Relative directory path to the manifest within the repo", default="") branch: Optional[str] = Field(description="Revision the user requested; empty means no explicit revision was requested. The fixed actually-cloned revision is baked into url; a bare stored URL means main.", default=None) + commit_sha: Optional[str] = Field( + description="Commit the app is pinned to; jobs run an immutable snapshot of this SHA. None for legacy rows not yet pinned.", + default=None, + ) + code_commit_sha: Optional[str] = Field( + description="Pin for the manifest's separate code repo (repo_url), when declared.", + default=None, + ) name: str = Field(description="App name from manifest") description: Optional[str] = Field(description="App description from manifest", default=None) added_at: datetime = Field(description="When the app was added") @@ -683,6 +818,15 @@ def validate_name(cls, v): class UpdateAppListingRequest(BaseModel): """Request to update a listing's editable metadata""" + url: Optional[str] = Field( + description=( + "New GitHub URL for the listing, optionally carrying a revision " + "as /tree/ (same form the add flow uses). When it differs " + "from the stored URL, the repo is cloned and the listing's " + "manifest path must still exist there." + ), + default=None, + ) name: Optional[str] = Field(description="New display name", default=None) description: Optional[str] = Field(description="New description", default=None) @@ -701,6 +845,40 @@ class ManifestFetchRequest(BaseModel): class AppAddRequest(BaseModel): """Request to add an app""" url: str = Field(description="URL to the app manifest or GitHub repo") + manifest_paths: Optional[List[str]] = Field( + description=( + "Manifest paths (relative dirs within the repo) to add. When omitted " + "or null, every discovered manifest in the repo is added. Use this to " + "add a subset of a multi-app repository." + ), + default=None, + ) + + +class DiscoveredApp(BaseModel): + """A manifest discovered in a repo, for previewing before adding""" + manifest_path: str = Field(description="Relative directory path to the manifest within the repo", default="") + name: str = Field(description="App name from the manifest") + description: Optional[str] = Field(description="App description from the manifest", default=None) + already_added: bool = Field( + description="True if the user has already added this manifest from this repo", + default=False, + ) + + +class AppUpdateCheck(BaseModel): + """Whether a newer commit exists on an app's pinned revision""" + url: str = Field(description="Canonical URL of the app") + manifest_path: str = Field(description="Manifest path within the repo", default="") + commit_sha: Optional[str] = Field(description="Commit the app is currently pinned to", default=None) + latest_sha: Optional[str] = Field( + description="Tip of the pinned revision on the remote; None if it could not be resolved", + default=None, + ) + update_available: bool = Field( + description="True when the remote tip differs from the pinned commit", + default=False, + ) class AppRemoveRequest(BaseModel): @@ -730,7 +908,7 @@ class Job(BaseModel): description="Environment-tab parameter values (separate namespace from parameters)", default=None, ) - status: str = Field(description="Job status (PENDING, RUNNING, DONE, FAILED, KILLED)") + status: str = Field(description="Job status (PENDING, RUNNING, UNKNOWN, DONE, FAILED, KILLED)") exit_code: Optional[int] = Field(description="Exit code of the job", default=None) resources: Optional[Dict] = Field(description="Requested resources", default=None) env: Optional[Dict[str, str]] = Field(description="Environment variables used for the job", default=None) @@ -742,8 +920,14 @@ class Job(BaseModel): conda_env: Optional[str] = Field(description="Conda environment activated for the job", default=None) requirements: Optional[List[str]] = Field(description="Declared runtime requirements (e.g. ['nextflow', 'apptainer'])", default=None) work_dir: Optional[str] = Field(description="Working directory the job ran in", default=None) + commit_sha: Optional[str] = Field(description="Commit whose code the job executed", default=None) + code_repo_url: Optional[str] = Field( + description="Repo the executed commit belongs to, when it differs from app_url", + default=None, + ) cluster_job_id: Optional[str] = Field(description="Cluster-assigned job ID", default=None) service_url: Optional[str] = Field(description="URL of the running service (for service-type jobs)", default=None) + phase: Optional[str] = Field(description="Startup phase of a running service before its URL is ready, e.g. 'pulling_image' or 'starting'", default=None) created_at: datetime = Field(description="When the job was created") started_at: Optional[datetime] = Field(description="When the job started running", default=None) finished_at: Optional[datetime] = Field(description="When the job finished", default=None) @@ -777,8 +961,19 @@ class JobSubmitRequest(BaseModel): @field_validator("extra_args") @classmethod def validate_extra_args(cls, v): - if v is not None and _SHELL_METACHAR_PATTERN.search(v): - raise ValueError(f"extra_args contains forbidden characters: {v!r}") + # extra_args are shlex-split into argv tokens and passed to the + # scheduler via exec (no shell — see cluster_api's bsub), so shell + # metacharacters are safe and in fact required: LSF resource strings + # like -R "select[mem>8000]" contain '>', '[' and ']'. Only require + # that the string parses into balanced tokens and carries no NUL. + if v is None: + return v + if "\x00" in v: + raise ValueError("extra_args must not contain NUL bytes") + try: + shlex.split(v) + except ValueError as e: + raise ValueError(f"extra_args could not be parsed into arguments: {e}") return v @field_validator("container") @@ -799,6 +994,16 @@ def validate_container_args(cls, v): class PathValidationRequest(BaseModel): """Request to validate file/directory paths""" paths: Dict[str, str] = Field(description="Map of parameter key to path value") + may_be_missing: List[str] = Field( + default=[], + description="Keys whose path may not exist yet (exists=false params): " + "validated for file-share containment only, not existence", + ) + types: Dict[str, str] = Field( + default={}, + description="Expected type per key ('file' or 'directory'): when the " + "path exists, its type must match", + ) class PathValidationResponse(BaseModel): diff --git a/fileglancer/server.py b/fileglancer/server.py index 6655a058..c9c9c4ab 100644 --- a/fileglancer/server.py +++ b/fileglancer/server.py @@ -1,5 +1,7 @@ +import asyncio import os import re +import shlex import sys try: import pwd @@ -527,8 +529,7 @@ def mask_password(url: str) -> str: logger.info("Worker pool started") # Wire the apps module to dispatch through the persistent worker - # pool (or in-process in dev mode) instead of spawning ephemeral - # subprocesses. + # pool (or in-process in dev mode). apps_module.set_worker_exec(_worker_exec) # Start cluster job monitor @@ -1680,6 +1681,8 @@ async def get_user_apps(username: str = Depends(get_current_user)): "name": row.name, "description": row.description, "branch": row.branch, + "commit_sha": row.commit_sha, + "code_commit_sha": row.code_commit_sha, "manifest": row.manifest, "added_at": row.added_at, "updated_at": row.updated_at, @@ -1710,6 +1713,8 @@ async def get_user_apps(username: str = Depends(get_current_user)): name=snap["name"], description=snap["description"], branch=snap["branch"], + commit_sha=snap["commit_sha"], + code_commit_sha=snap["code_commit_sha"], added_at=snap["added_at"], updated_at=snap["updated_at"], manifest=manifest_obj, @@ -1726,21 +1731,78 @@ async def get_user_apps(username: str = Depends(get_current_user)): logger.warning(f"Failed to fetch manifest for {snap['url']}: {e}") continue - user_app = result[idx] - user_app.manifest = manifest - user_app.name = manifest.name - user_app.description = manifest.description + # Only the manifest is refreshed; name/description keep the row's + # values, which may be user-chosen (e.g. a custom catalog name). + result[idx].manifest = manifest return result - @app.post("/api/apps", response_model=list[UserApp], - description="Add an app by URL (discovers all manifests in the repo)") - async def add_user_app(body: AppAddRequest, - username: str = Depends(get_current_user)): - # Clone the repo and discover all manifests. The worker resolves the - # branch as the user, so a private repo's real default is used. + # How long a finished job keeps its snapshot alive. Work dirs carry a + # `repo` symlink into the snapshot, so collecting it immediately would + # break browsing a recent job's code; jobs older than this trade that + # for reclaiming the (potentially large) snapshot tree. + _JOB_SNAPSHOT_RETENTION = timedelta(days=14) + + def _collect_keep_shas(session, username: str) -> list[str]: + """Every snapshot SHA still referenced by this user: all app pins, + plus the commits of jobs that are not in a terminal state (UNKNOWN + counts as live — the poll loop writes raw executor statuses), plus + recently-created terminal jobs (their work dirs link into the + snapshot).""" + keep = set() + for row in db.list_user_apps(session, username): + if row.commit_sha: + keep.add(row.commit_sha) + if row.code_commit_sha: + keep.add(row.code_commit_sha) + cutoff = datetime.now(UTC).replace(tzinfo=None) - _JOB_SNAPSHOT_RETENTION + for j in db.get_jobs_by_username(session, username): + if not j.commit_sha: + continue + created = j.created_at.replace(tzinfo=None) if j.created_at else None + if not db.is_terminal_job_status(j.status) or ( + created is not None and created >= cutoff): + keep.add(j.commit_sha) + return sorted(keep) + + # Fire-and-forget tasks must be referenced until done — the event loop + # only holds weak references, so an unreferenced task can be gc'd + # mid-execution. + _background_tasks: set = set() + + def _spawn_snapshot_gc(username: str, urls: list) -> None: + task = asyncio.create_task(_gc_app_snapshots(username, urls)) + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) + + async def _gc_app_snapshots(username: str, urls: list): + """Fire-and-forget snapshot GC for the given repos. + + The keep-set is over-inclusive (all of the user's pins, not just this + repo's) — a SHA that doesn't exist under a repo's .snapshots simply + matches nothing, and over-keeping is always safe. + """ try: - resolved_branch, discovered = await apps_module.discover_app_manifests( - body.url, username=username) + with db.get_db_session(settings.db_url) as session: + keep = _collect_keep_shas(session, username) + except Exception as e: + logger.warning(f"Snapshot GC skipped for {username}: {e}") + return + for url in dict.fromkeys(u for u in urls if u): + try: + await apps_module.gc_repo_snapshots(url, keep, username=username) + except Exception as e: + logger.warning(f"Snapshot GC failed for {url} ({username}): {e}") + + async def _discover_repo_manifests(url: str, username: str): + """Clone/scan a repo and return (resolved_branch, head_sha, + canonical_url, discovered). + + Shared by the discover and add endpoints so both surface identical, + user-facing errors for a bad URL/revision/private-repo clone. + """ + try: + resolved_branch, head_sha, discovered = await apps_module.discover_app_manifests( + url, username=username) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except HTTPException as e: @@ -1764,9 +1826,50 @@ async def add_user_app(body: AppAddRequest, ) # Bake the worker-resolved revision into the stored URL (so a repo whose - # default is e.g. "master" dedups against an explicit ".../tree/master"), - # and record the user's requested revision separately ("" = unpinned). - canonical_url, requested = apps_module.canonical_app_url(body.url, resolved_branch) + # default is e.g. "master" dedups against an explicit ".../tree/master"). + canonical_url, _ = apps_module.canonical_app_url(url, resolved_branch) + return resolved_branch, head_sha, canonical_url, discovered + + @app.post("/api/apps/discover", response_model=list[DiscoveredApp], + description="Discover the apps (manifests) in a repo without adding them") + async def discover_user_apps(body: AppAddRequest, + username: str = Depends(get_current_user)): + _, _, canonical_url, discovered = await _discover_repo_manifests(body.url, username) + with db.get_db_session(settings.db_url) as session: + return [ + DiscoveredApp( + manifest_path=manifest_path, + name=manifest.name, + description=manifest.description, + already_added=db.get_user_app( + session, username, canonical_url, manifest_path) is not None, + ) + for manifest_path, manifest in discovered + ] + + @app.post("/api/apps", response_model=list[UserApp], + description="Add apps by URL (all discovered manifests, or the subset in manifest_paths)") + async def add_user_app(body: AppAddRequest, + username: str = Depends(get_current_user)): + # Clone the repo and discover all manifests. The worker resolves the + # branch as the user, so a private repo's real default is used. + resolved_branch, head_sha, canonical_url, discovered = await _discover_repo_manifests( + body.url, username) + + # Restrict to the requested subset when manifest_paths is provided; an + # omitted/null list means "add every discovered manifest". Paths not + # present in the repo are ignored. + if body.manifest_paths is not None: + wanted = set(body.manifest_paths) + discovered = [(p, m) for p, m in discovered if p in wanted] + if not discovered: + raise HTTPException( + status_code=400, + detail="None of the requested apps were found in the repository.", + ) + + # Record the user's requested revision separately ("" = unpinned). + _, requested = apps_module.canonical_app_url(body.url, resolved_branch) new_apps: list[UserApp] = [] with db.get_db_session(settings.db_url) as session: @@ -1778,12 +1881,15 @@ async def add_user_app(body: AppAddRequest, url=canonical_url, manifest_path=manifest_path, name=manifest.name, description=manifest.description, branch=requested, + commit_sha=head_sha, manifest=manifest.model_dump(mode="json"), ) new_apps.append(UserApp( url=row.url, manifest_path=row.manifest_path, branch=row.branch, + commit_sha=row.commit_sha, + code_commit_sha=row.code_commit_sha, name=row.name, description=row.description, added_at=row.added_at, @@ -1797,6 +1903,40 @@ async def add_user_app(body: AppAddRequest, detail="All apps in this repository have already been added.", ) + # Materialize the pinned snapshot now so the first launch doesn't pay + # for it. Non-fatal: launch re-ensures the snapshot if this failed. + if head_sha: + try: + clone_url = apps_module.clone_url_for_stored_app(canonical_url, requested) + await apps_module.ensure_repo_snapshot( + clone_url, sha=head_sha, username=username) + except Exception as e: + logger.warning(f"Eager snapshot of {canonical_url}@{head_sha[:7]} failed: {e}") + + # Pin each app's separate code repo (manifest repo_url) now, at add + # time, so a later launch can't silently run code that moved after the + # app was added/reviewed — and so update-checks can see code drift + # before the first launch. Best-effort: submit_job still backfills the + # code pin at launch if this fails (e.g. a transient network error), + # so a flaky code remote never blocks adding the app. + for app in new_apps: + repo_url = app.manifest.repo_url if app.manifest else None + if not repo_url or canonical_github_url(repo_url) == canonical_url: + continue + try: + _, code_sha = await apps_module.ensure_repo_snapshot( + repo_url, pull=True, username=username) + except Exception as e: + logger.warning( + f"Eager code-repo pin of {repo_url} for {canonical_url} " + f"({app.manifest_path}) failed: {e}") + continue + with db.get_db_session(settings.db_url) as session: + db.set_user_app_pins( + session, username, canonical_url, app.manifest_path, + code_commit_sha=code_sha) + app.code_commit_sha = code_sha + return new_apps @app.delete("/api/apps", @@ -1805,41 +1945,55 @@ async def remove_user_app(url: str = Query(..., description="URL of the app to r manifest_path: str = Query("", description="Manifest path within the repo"), username: str = Depends(get_current_user)): with db.get_db_session(settings.db_url) as session: - if not db.delete_user_app(session, username, url, manifest_path): + row = db.get_user_app(session, username, url, manifest_path) + if row is None: raise HTTPException(status_code=404, detail="App not found") + # Repos whose snapshots may have just become unreferenced: the + # app's own repo, plus its separate code repo if one was declared. + gc_urls = [row.url, (row.manifest or {}).get("repo_url")] + db.delete_user_app(session, username, url, manifest_path) + _spawn_snapshot_gc(username, gc_urls) return {"message": "App removed"} @app.post("/api/apps/update", response_model=UserApp, - description="Pull latest code and re-read the manifest for an app") + description="Pull latest code and re-pin this app to the new tip commit") async def update_user_app(body: ManifestFetchRequest, username: str = Depends(get_current_user)): # The revision is fixed at add time and baked into body.url, so update - # just pulls that revision again and re-reads the manifest — it never - # re-resolves the default branch or moves the app to a new URL. + # just pulls that revision again — it never re-resolves the default + # branch or moves the app to a new URL. Only THIS app's pin moves: + # sibling apps in the same repo keep their snapshots, and running jobs + # keep the tree they started with. with db.get_db_session(settings.db_url) as session: existing = db.get_user_app(session, username, body.url, body.manifest_path) if existing is None: raise HTTPException(status_code=404, detail="App not found") stored_url = existing.url stored_branch = existing.branch + # If the update changes the manifest's repo_url, the old code + # repo's snapshots lose their last reference here — remember it + # so the GC below can target it. + old_repo_url = (existing.manifest or {}).get("repo_url") clone_url = apps_module.clone_url_for_stored_app(stored_url, stored_branch) try: - await apps_module._ensure_repo_cache(clone_url, pull=True, username=username) + _, new_sha = await apps_module.ensure_repo_snapshot( + clone_url, pull=True, username=username) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to pull latest code: {str(e)}") try: manifest = await apps_module.fetch_app_manifest(clone_url, body.manifest_path, - username=username) + username=username, sha=new_sha) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to read manifest after update: {str(e)}") + code_sha = None if manifest.repo_url and canonical_github_url(manifest.repo_url) != stored_url: try: - await apps_module._ensure_repo_cache( + _, code_sha = await apps_module.ensure_repo_snapshot( manifest.repo_url, pull=True, username=username, @@ -1856,24 +2010,91 @@ async def update_user_app(body: ManifestFetchRequest, session, username, url=stored_url, manifest_path=body.manifest_path, name=manifest.name, description=manifest.description, + commit_sha=new_sha, code_commit_sha=code_sha, manifest=manifest.model_dump(mode="json"), ) - return UserApp( + result = UserApp( url=row.url, manifest_path=row.manifest_path, branch=row.branch, + commit_sha=row.commit_sha, + code_commit_sha=row.code_commit_sha, name=row.name, description=row.description, added_at=row.added_at, updated_at=row.updated_at, manifest=manifest, ) + _spawn_snapshot_gc(username, [stored_url, manifest.repo_url, old_repo_url]) + return result + + @app.get("/api/apps/check-updates", response_model=list[AppUpdateCheck], + description="Compare each app's pinned commits against their remote revision tips") + async def check_app_updates(username: str = Depends(get_current_user)): + with db.get_db_session(settings.db_url) as session: + rows = [ + { + "url": r.url, + "manifest_path": r.manifest_path, + "branch": r.branch, + "commit_sha": r.commit_sha, + "code_commit_sha": r.code_commit_sha, + "repo_url": (r.manifest or {}).get("repo_url"), + } + for r in db.list_user_apps(session, username) + ] + + # One batched worker round-trip resolves every distinct repo+revision + # (sibling apps share a lookup, and the ls-remotes run concurrently in + # the worker, so a slow remote costs one short timeout — not one per + # app queued on the user's serial worker). + lookup_urls: list[str] = [] + for row in rows: + if not row["commit_sha"]: + continue + row["clone_url"] = apps_module.clone_url_for_stored_app( + row["url"], row["branch"]) + lookup_urls.append(row["clone_url"]) + if row["code_commit_sha"] and row["repo_url"]: + lookup_urls.append(row["repo_url"]) + tips: dict[str, Optional[str]] = {} + if lookup_urls: + try: + tips = await apps_module.get_remote_heads( + lookup_urls, username=username) + except Exception as e: + logger.warning(f"Update check failed for {username}: {e}") + + results: list[AppUpdateCheck] = [] + for row in rows: + if not row["commit_sha"]: + # Unpinned legacy row — nothing to compare against. + results.append(AppUpdateCheck( + url=row["url"], manifest_path=row["manifest_path"])) + continue + latest = tips.get(row["clone_url"]) + drifted = bool(latest) and latest != row["commit_sha"] + # An app whose code lives in a separate repo can also drift there. + if row["code_commit_sha"] and row["repo_url"]: + code_latest = tips.get(row["repo_url"]) + drifted = drifted or ( + bool(code_latest) and code_latest != row["code_commit_sha"]) + results.append(AppUpdateCheck( + url=row["url"], + manifest_path=row["manifest_path"], + commit_sha=row["commit_sha"], + latest_sha=latest, + update_available=drifted, + )) + return results @app.post("/api/apps/validate-paths", response_model=PathValidationResponse, description="Validate file/directory paths for app parameters") async def validate_paths(body: PathValidationRequest, username: str = Depends(get_current_user)): - result = await _worker_exec(username, "validate_paths", paths=body.paths) + result = await _worker_exec(username, "validate_paths", paths=body.paths, + may_be_missing=body.may_be_missing, + types=body.types) return PathValidationResponse(errors=result.get("errors", {})) # --- Catalog (shared apps) API --- @@ -1934,10 +2155,43 @@ async def update_catalog_listing(listing_id: int, body: UpdateAppListingRequest, username: str = Depends(get_current_user)): with db.get_db_session(settings.db_url) as session: - listing = db.update_app_listing( - session, listing_id, username, - name=body.name, description=body.description, - ) + existing = db.get_app_listing(session, listing_id) + if existing is None or existing.owner_username != username: + raise HTTPException(status_code=404, detail="Listing not found") + current_url = existing.url + manifest_path = existing.manifest_path + + # Repointing the listing at a different repo/revision: clone/scan it + # as the user (outside any DB session — this can take a while) and + # require the listing's manifest path to still exist there, so the + # catalog never advertises an app that can't be added. + new_url = None + new_branch = None + if body.url is not None and canonical_github_url(body.url) != current_url: + resolved_branch, _, canonical_url, discovered = await _discover_repo_manifests( + body.url, username) + found_paths = [p for p, _ in discovered] + if manifest_path not in found_paths: + locations = ", ".join(f"'{p}'" if p else "the repository root" + for p in found_paths) + raise HTTPException( + status_code=400, + detail=f"No app manifest found at '{manifest_path or '(root)'}' " + f"in that repository/revision. Manifests were found at: " + f"{locations}.", + ) + new_url = canonical_url + _, new_branch = apps_module.canonical_app_url(body.url, resolved_branch) + + with db.get_db_session(settings.db_url) as session: + try: + listing = db.update_app_listing( + session, listing_id, username, + name=body.name, description=body.description, + url=new_url, branch=new_branch, + ) + except ValueError as e: + raise HTTPException(status_code=409, detail=str(e)) if listing is None: raise HTTPException(status_code=404, detail="Listing not found") return _listing_to_model(listing) @@ -1973,14 +2227,39 @@ async def add_from_catalog(listing_id: int, clone_url = apps_module.clone_url_for_stored_app(listing_url, listing_branch) try: + # Pin the install to the revision's current tip and read the + # manifest from that immutable snapshot. + try: + _, pinned_sha = await apps_module.ensure_repo_snapshot( + clone_url, pull=True, username=username) + except Exception: + # The pull needs the network; a warm cached clone doesn't. + # Fall back so installing a previously-cloned app still works + # offline (pinned to the cache's tip instead of the remote's). + _, pinned_sha = await apps_module.ensure_repo_snapshot( + clone_url, username=username) manifest = await apps_module.fetch_app_manifest( - clone_url, listing_manifest_path, username=username, + clone_url, listing_manifest_path, username=username, sha=pinned_sha, ) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to fetch manifest: {str(e)}") + # Pin the manifest's separate code repo (repo_url) now, at add time, so + # a later launch can't silently run code that moved after the app was + # added. Best-effort: submit_job backfills this pin at launch if it + # fails here, so a flaky code remote never blocks adding the app. + code_sha = None + if manifest.repo_url and canonical_github_url(manifest.repo_url) != listing_url: + try: + _, code_sha = await apps_module.ensure_repo_snapshot( + manifest.repo_url, pull=True, username=username) + except Exception as e: + logger.warning( + f"Eager code-repo pin of {manifest.repo_url} for " + f"{listing_url} ({listing_manifest_path}) failed: {e}") + # The listing already carries the canonical URL (resolved revision baked # in) and the requested revision, so copy them straight over. with db.get_db_session(settings.db_url) as session: @@ -1989,12 +2268,16 @@ async def add_from_catalog(listing_id: int, url=listing_url, manifest_path=listing_manifest_path, name=listing_name, description=listing_description, branch=listing_branch, + commit_sha=pinned_sha, + code_commit_sha=code_sha, manifest=manifest.model_dump(mode="json"), ) return UserApp( url=row.url, manifest_path=row.manifest_path, branch=row.branch, + commit_sha=row.commit_sha, + code_commit_sha=row.code_commit_sha, name=row.name, description=row.description, added_at=row.added_at, @@ -2006,7 +2289,7 @@ async def add_from_catalog(listing_id: int, description="Get cluster configuration defaults") async def get_cluster_defaults(): return { - "extra_args": " ".join(settings.cluster.extra_args), + "extra_args": shlex.join(settings.cluster.extra_args), } @app.post("/api/jobs", response_model=Job, @@ -2033,6 +2316,10 @@ async def submit_job(body: JobSubmitRequest, container=body.container, container_args=body.container_args, ) + # Launches of apps not in the user's library create snapshots that + # no delete/update endpoint ever GCs; sweep here (the new job's + # row is committed, so its own snapshot is in the keep-set). + _spawn_snapshot_gc(username, [body.app_url]) return _convert_job(db_job) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @@ -2046,17 +2333,36 @@ async def get_jobs(status: Optional[str] = Query(None, description="Filter by st username: str = Depends(get_current_user)): with db.get_db_session(settings.db_url) as session: db_jobs = db.get_jobs_by_username(session, username, status) - # For listing, read service_url for running service jobs via worker + # Read service_url/phase for all running service jobs in a single + # worker round-trip. This endpoint is polled every few seconds and + # the worker executes one action at a time, so dispatching per job + # would serially block the worker (and the user's file browsing) + # once several services are running. + running_services = [ + j for j in db_jobs + if getattr(j, 'entry_point_type', 'job') == 'service' + and j.status == 'RUNNING' + ] + services = {} + if running_services: + try: + result = await _worker_exec( + username, "get_service_urls", + jobs=[{ + "id": j.id, + "app_name": j.app_name, + "entry_point_id": j.entry_point_id, + "work_dir": j.work_dir, + } for j in running_services], + ) + services = result.get("services") or {} + except Exception: + pass jobs = [] for j in db_jobs: - service_url = None - if getattr(j, 'entry_point_type', 'job') == 'service' and j.status == 'RUNNING': - try: - result = await _worker_exec(username, "get_service_url", job_id=j.id) - service_url = result.get("service_url") - except Exception: - pass - jobs.append(_convert_job(j, service_url=service_url)) + info = services.get(str(j.id)) or {} + jobs.append(_convert_job(j, service_url=info.get("service_url"), + phase=info.get("phase"))) return JobResponse(jobs=jobs) @app.get("/api/jobs/{job_id}", response_model=Job, @@ -2072,13 +2378,15 @@ async def get_job(job_id: int, # rather than paying a worker roundtrip + NFS glob/stat. files = apps_module.get_job_file_paths(db_job) service_url = None + phase = None if getattr(db_job, 'entry_point_type', 'job') == 'service' and db_job.status == 'RUNNING': try: svc_result = await _worker_exec(username, "get_service_url", job_id=job_id) service_url = svc_result.get("service_url") + phase = svc_result.get("phase") except Exception: pass - return _convert_job(db_job, service_url=service_url, files=files) + return _convert_job(db_job, service_url=service_url, files=files, phase=phase) @app.post("/api/jobs/{job_id}/cancel", description="Cancel a running job") @@ -2091,18 +2399,24 @@ async def cancel_job(job_id: int, raise HTTPException(status_code=400, detail=str(e)) @app.delete("/api/jobs/{job_id}", - description="Delete a job record") + description="Delete a job record and its work directory") async def delete_job(job_id: int, username: str = Depends(get_current_user)): with db.get_db_session(settings.db_url) as session: db_job = db.get_job(session, job_id, username) if db_job is None: raise HTTPException(status_code=404, detail="Job not found") - if db_job.status in ("PENDING", "RUNNING"): + if not db.is_terminal_job_status(db_job.status): raise HTTPException( status_code=409, detail="Job is active; cancel or stop it before deleting.", ) + result = await _worker_exec(username, "delete_job_work_dir", job_id=job_id) + if result.get("error"): + raise HTTPException( + status_code=result.get("status_code", 500), + detail=result["error"], + ) db.delete_job(session, job_id, username) return {"message": "Job deleted"} @@ -2139,11 +2453,12 @@ def _ensure_utc(dt: Optional[datetime]) -> Optional[datetime]: return dt.replace(tzinfo=UTC) return dt - def _convert_job(db_job: db.JobDB, service_url: str = None, files: dict = None) -> Job: + def _convert_job(db_job: db.JobDB, service_url: str = None, files: dict = None, + phase: str = None) -> Job: """Convert a database JobDB to a Pydantic Job model. - File-reading fields (service_url, files) must be passed in pre-computed - by the caller, since they require user-context file I/O. + File-reading fields (service_url, phase, files) must be passed in + pre-computed by the caller, since they require user-context file I/O. """ return Job( id=db_job.id, @@ -2167,8 +2482,11 @@ def _convert_job(db_job: db.JobDB, service_url: str = None, files: dict = None) conda_env=db_job.conda_env, requirements=db_job.requirements, work_dir=db_job.work_dir, + commit_sha=db_job.commit_sha, + code_repo_url=db_job.code_repo_url, cluster_job_id=db_job.cluster_job_id, service_url=service_url, + phase=phase, created_at=_ensure_utc(db_job.created_at), started_at=_ensure_utc(db_job.started_at), finished_at=_ensure_utc(db_job.finished_at), diff --git a/fileglancer/settings.py b/fileglancer/settings.py index 9563cba8..68c8ba4f 100644 --- a/fileglancer/settings.py +++ b/fileglancer/settings.py @@ -38,6 +38,19 @@ class ClusterSettings(BaseModel): class AppsSettings(BaseModel): """Apps-specific configuration (not passed to py-cluster-api).""" extra_paths: List[str] = [] + # Extra environment variable names (exact) or prefixes (ending in '_') to + # pass through to per-user workers, on top of the built-in allowlist. The + # worker env is an allowlist so no server secret leaks to the user via + # /proc//environ; add site-specific vars your scheduler or tools need + # here rather than widening the allowlist in code. FGC_* is never passed + # through regardless of this list. + worker_env_passthrough: List[str] = [] + # How long a job may sit in UNKNOWN before the poll loop gives up and marks + # it FAILED. UNKNOWN means the scheduler can no longer report the job (it + # aged out of the queue/history), so continued polling would never resolve + # it. This is Fileglancer poll policy, not py-cluster-api config, so it + # lives here rather than under `cluster`. Set to 0 to disable the cutoff. + unknown_timeout_hours: float = 24.0 class Settings(BaseSettings): diff --git a/fileglancer/user_worker.py b/fileglancer/user_worker.py index 34112abe..95f42a34 100644 --- a/fileglancer/user_worker.py +++ b/fileglancer/user_worker.py @@ -36,6 +36,7 @@ import socket import struct import sys +import time from pathlib import Path from typing import Any, Optional @@ -167,8 +168,8 @@ def _job_db_to_dict(j) -> dict: """Serialize a JobDB row to a JSON-safe dict for transport to the worker. Only includes fields used by worker-side handlers (read_job_file, - get_service_url) — keep this list minimal so the worker sees as little of - the DB row as possible. + get_service_url, delete_job_work_dir) — keep this list minimal so the + worker sees as little of the DB row as possible. """ return { "id": j.id, @@ -646,15 +647,57 @@ def _action_validate_paths(request: dict, ctx: WorkerContext) -> dict: from fileglancer.apps.command import validate_path_in_filestore paths = request["paths"] + # Keys whose path may not exist yet (exists=false params) are checked for + # file-share containment only — these are outputs the job creates + # (directories are created by Fileglancer at submit time), so a + # missing-but-in-share path is valid here. + may_be_missing = set(request.get("may_be_missing") or []) + # Expected type per key ('file' or 'directory'): when the path exists, its + # type must match. + types = request.get("types") or {} fsps = ctx.db.get_file_share_paths() errors = {} for param_key, path_value in paths.items(): - error = validate_path_in_filestore(path_value, fsps) + error = validate_path_in_filestore( + path_value, fsps, check_access=param_key not in may_be_missing, + expected_type=types.get(param_key) + ) if error: errors[param_key] = error return {"errors": errors} +@action("create_dirs") +def _action_create_dirs(request: dict, ctx: WorkerContext) -> dict: + """Create directories for app params with exists=false. + + Runs as the target user in the setuid worker. Each path is expanded ('~' + resolves to this user's home), confirmed to be within an allowed file share + (containment only — the directory need not exist yet), and only then created + with exist_ok=True. Never creates anything outside a file share. Best-effort: + per-path failures are returned in {"errors": {index: message}} rather than + aborting the batch. + """ + from fileglancer.apps.command import validate_path_in_filestore + + paths = request["paths"] + fsps = ctx.db.get_file_share_paths() + errors = {} + for key, path_value in paths.items(): + # Containment check without the exists/readable check — the directory is + # about to be created, so it legitimately may not exist yet. + error = validate_path_in_filestore(path_value, fsps, check_access=False) + if error: + errors[key] = error + continue + expanded = os.path.expanduser(path_value.replace("\\", "/")) + try: + os.makedirs(expanded, exist_ok=True) + except OSError as e: + errors[key] = str(e) + return {"errors": errors} + + @action("get_profile") def _action_get_profile(request: dict, ctx: WorkerContext) -> dict: """Get user profile information.""" @@ -741,17 +784,196 @@ def _action_get_job_file(request: dict, ctx: WorkerContext) -> dict: return {"content": content} +@action("delete_job_work_dir") +def _action_delete_job_work_dir(request: dict, ctx: WorkerContext) -> dict: + """Delete a terminal job's work directory as the target user.""" + from fileglancer import database as db + from fileglancer.apps.jobfiles import delete_job_work_dir + + job_id = request["job_id"] + db_job = ctx.db.get_job(job_id, ctx.username) + if db_job is None: + return {"error": f"Job {job_id} not found", "status_code": 404} + if not db.is_terminal_job_status(db_job.status): + return { + "error": "Job is active; cancel or stop it before deleting.", + "status_code": 409, + } + + try: + deleted = delete_job_work_dir(db_job) + return {"ok": True, "work_dir_deleted": deleted} + except PermissionError as e: + return {"error": str(e), "status_code": 403} + except OSError as e: + return {"error": str(e), "status_code": 500} + + +def _proc_alive(pid: int) -> bool: + """True if pid is a live (non-zombie) process. + + Uses /proc so a killed-but-not-yet-reaped process (state 'Z') counts as + dead — os.kill(pid, 0) alone would report a zombie as alive. + """ + try: + with open(f"/proc/{pid}/stat", "rb") as f: + data = f.read() + # "pid (comm) state ...": comm may contain spaces/parens, so read the + # state character from just after the final ')'. + state = data[data.rindex(b")") + 2:data.rindex(b")") + 3] + return state not in (b"Z", b"X", b"x") + except (FileNotFoundError, ProcessLookupError): + return False + except OSError: + # /proc unavailable — fall back to a signal-0 probe. + try: + os.kill(pid, 0) + return True + except ProcessLookupError: + return False + except PermissionError: + return True + + +def _descendant_pids(root_pid: int) -> list[int]: + """Return all descendant PIDs of root_pid (children, grandchildren, ...). + + Reads PPID from /proc//stat. Must be called before the root is killed: + once a parent dies its children reparent to init and can no longer be + traced back to root_pid. + """ + children: dict[int, list[int]] = {} + for entry in os.listdir("/proc"): + if not entry.isdigit(): + continue + try: + with open(f"/proc/{entry}/stat", "rb") as f: + data = f.read() + fields = data[data.rindex(b")") + 2:].split() + ppid = int(fields[1]) # state, ppid, ... + except (OSError, ValueError, IndexError): + continue + children.setdefault(ppid, []).append(int(entry)) + + result: list[int] = [] + stack = [root_pid] + while stack: + for child in children.get(stack.pop(), []): + result.append(child) + stack.append(child) + return result + + +def _terminate_process_tree(pid: int, grace_seconds: float = 3.0) -> bool: + """Terminate pid and all its descendants; return True once none survive. + + The launcher bash does not forward SIGTERM to its foreground child, so + signalling the launcher alone leaves the real workload running (orphaned). + Snapshot the whole tree first (before anything dies and children reparent), + SIGTERM it, then SIGKILL any that outlast the grace period. + """ + import signal + + # Snapshot children before signalling; include the launcher itself. + targets = _descendant_pids(pid) + [pid] + + def _signal_all(sig: int) -> None: + for target in targets: + try: + os.kill(target, sig) + except (ProcessLookupError, PermissionError): + pass + + def _any_alive() -> bool: + return any(_proc_alive(target) for target in targets) + + _signal_all(signal.SIGTERM) + deadline = time.monotonic() + grace_seconds + while _any_alive() and time.monotonic() < deadline: + time.sleep(0.1) + + if _any_alive(): + _signal_all(signal.SIGKILL) + # SIGKILL is immediate but reaping isn't; give the kernel a moment. + time.sleep(0.2) + + return not _any_alive() + + +@action("cancel_local") +def _action_cancel_local(request: dict, ctx: WorkerContext) -> dict: + """Terminate a local-executor job and its whole process tree, as the user. + + The local executor spawns `bash