app.py

"""Minimal bucket mount probe for Gradio SDK Spaces.

Runs the probe once at import time (so results appear in startup logs even if
the Gradio UI fails to load), and also exposes it via a Gradio interface.
"""

import os
import pwd
import sqlite3
import subprocess
import traceback

try:
    import fcntl
except ImportError:
    fcntl = None

import gradio as gr


def _run_probe() -> str:
    lines = []
    lines.append("=" * 60)
    lines.append("BUCKET MOUNT PROBE — Gradio SDK Space")
    lines.append("=" * 60)

    # Runtime identity
    try:
        uid = os.getuid()
        user = pwd.getpwuid(uid).pw_name
        lines.append(f"uid={uid} user={user} euid={os.geteuid()} gid={os.getgid()}")
    except Exception as e:
        lines.append(f"uid/user lookup failed: {e}")

    # Env hints
    lines.append(f"SYSTEM={os.environ.get('SYSTEM')!r}")
    lines.append(f"SPACE_ID={os.environ.get('SPACE_ID')!r}")
    lines.append(f"HOME={os.environ.get('HOME')!r}")

    # Mount listing
    lines.append("")
    lines.append("--- ls -lan /data ---")
    try:
        out = subprocess.run(
            ["ls", "-lan", "/data"], capture_output=True, text=True, timeout=10
        )
        lines.append(out.stdout or "(empty)")
        if out.stderr:
            lines.append(f"stderr: {out.stderr}")
    except Exception as e:
        lines.append(f"ls failed: {e}")

    lines.append("--- stat /data ---")
    try:
        out = subprocess.run(
            ["stat", "/data"], capture_output=True, text=True, timeout=10
        )
        lines.append(out.stdout or "(empty)")
    except Exception as e:
        lines.append(f"stat failed: {e}")

    lines.append("--- mount | grep /data ---")
    try:
        out = subprocess.run(
            "mount | grep /data || true",
            shell=True,
            capture_output=True,
            text=True,
            timeout=10,
        )
        lines.append(out.stdout or "(no match)")
    except Exception as e:
        lines.append(f"mount failed: {e}")

    # Write probes
    lines.append("")
    lines.append("--- write probes ---")

    def probe(label: str, fn):
        try:
            fn()
            lines.append(f"OK   {label}")
        except Exception as e:
            lines.append(f"FAIL {label}: {type(e).__name__}: {e}")

    probe(
        "touch /data/gradio_probe_touch",
        lambda: open("/data/gradio_probe_touch", "w").close(),
    )

    def _sqlite_create():
        conn = sqlite3.connect("/data/gradio_probe.db", timeout=5.0)
        conn.execute("CREATE TABLE IF NOT EXISTS t(x INTEGER)")
        conn.execute("INSERT INTO t VALUES (1)")
        conn.commit()
        conn.close()

    probe("sqlite3 connect + CREATE + INSERT", _sqlite_create)

    def _sqlite_delete_journal():
        # This is what trackio does on Spaces: journal_mode=DELETE instead of WAL
        conn = sqlite3.connect("/data/gradio_probe_delete.db", timeout=5.0)
        conn.execute("PRAGMA journal_mode = DELETE")
        conn.execute("CREATE TABLE IF NOT EXISTS t(x INTEGER)")
        conn.commit()
        conn.close()

    probe("sqlite3 journal_mode=DELETE", _sqlite_delete_journal)

    if fcntl is not None:

        def _flock():
            f = open("/data/gradio_probe.lock", "w")
            fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
            f.close()

        probe("fcntl.flock LOCK_EX|LOCK_NB", _flock)
    else:
        lines.append("SKIP fcntl.flock: fcntl not importable")

    lines.append("=" * 60)
    return "\n".join(lines)


# Run once at import time so results land in startup logs
try:
    _STARTUP_PROBE = _run_probe()
except Exception:
    _STARTUP_PROBE = "startup probe crashed:\n" + traceback.format_exc()

print(_STARTUP_PROBE, flush=True)


def probe_handler():
    return _run_probe()


with gr.Blocks() as demo:
    gr.Markdown("# Bucket mount probe — Gradio SDK Space")
    gr.Markdown(
        "Attach a Storage Bucket at `/data` via Space settings (Read/Write), "
        "then click the button below. Startup-time probe output is in the Space logs."
    )
    with gr.Row():
        btn = gr.Button("Run probe")
    out = gr.Textbox(label="Probe output", lines=30, value=_STARTUP_PROBE)
    btn.click(fn=probe_handler, inputs=[], outputs=out)

demo.launch()