bug(import): Cannot resume import after server crash - stale state blocks recovery

[caterryan]

Summary

After server crash, orphaned imports show "already running" and cannot be resumed. Breaks crash recovery.

Steps to Reproduce

1. Start an import with slow query interval (query_interval_ms: 15000)
2. While import is running, kill the server (pkill -9 influxdb3)
3. Restart the server
4. Check status - shows overall_status: "running" but nothing is running
5. Attempt to resume - receive error: "Import is already running"

Expected Behavior

After server restart, resume should either:

• Detect the import is not actually running and allow resume
• Automatically detect stale in_progress state and restart from checkpoint (or beginning if no checkpoint)

Actual Behavior

Resume is blocked with "already running" error. Import is stuck forever.

[caterryan]

Additional ai generated context:

Root Cause

The get_import_pause_state() function only checks the import_pause_state table:

def get_import_pause_state(influxdb3_local, import_id, task_id) -> ImportPauseState:
    # Queries import_pause_state table
    # Returns RUNNING if paused=false AND canceled=false

After a crash:

• import_pause_state has paused=false, canceled=false → returns RUNNING
• import_state has status="in_progress" with no paused_at_time
• No actual process is running the import

The system cannot distinguish between:

Scenario	import_pause_state	import_state	Actual State
Active import	paused=false	in_progress	HTTP request blocked, actively importing
Crashed import	paused=false	in_progress	No process running, orphaned state

Proposed Fix

Option A: Heartbeat-based detection (Recommended)

Add a heartbeat mechanism to detect stale imports:

1. Add last_heartbeat field to import_state or a new import_heartbeat table
2. Update heartbeat every batch (or every N seconds)
3. In resume_import(), check if in_progress but heartbeat is stale (>60s old)
4. If stale, treat as crashed and allow resume

HEARTBEAT_STALE_THRESHOLD_SECONDS = 60

def is_import_actually_running(influxdb3_local, import_id, task_id) -> bool:
    """Check if import is actively running vs orphaned/crashed"""
    query = f"""
    SELECT last_heartbeat
    FROM import_heartbeat
    WHERE import_id = '{import_id}'
    ORDER BY time DESC
    LIMIT 1
    """
    result = influxdb3_local.query(query)

    if not result:
        return False  # No heartbeat = not running

    last_heartbeat = result[0].get("last_heartbeat")
    if last_heartbeat is None:
        return False

    # Check if heartbeat is recent
    now = time.time()
    heartbeat_age = now - (last_heartbeat / 1e9)  # Convert ns to seconds

    return heartbeat_age < HEARTBEAT_STALE_THRESHOLD_SECONDS

2. Update import_table() to write heartbeat every batch:

def import_table(...):
    # ... in the import loop ...

    # Write heartbeat
    heartbeat_builder = LineBuilder("import_heartbeat")
    heartbeat_builder.tag("import_id", import_id)
    heartbeat_builder.int64_field("last_heartbeat", int(time.time() * 1e9))
    heartbeat_builder.time_ns(int(time.time() * 1e9))
    influxdb3_local.write_sync(heartbeat_builder, no_sync=True)

3. Update resume_import() to use heartbeat check:

def resume_import(...):
    pause_state = get_import_pause_state(...)

    if pause_state == ImportPauseState.RUNNING:
        # Check if actually running or just stale
        if is_import_actually_running(influxdb3_local, import_id, task_id):
            return {"status": "error", "error": f"Import {import_id} is already running"}
        else:
            influxdb3_local.info(f"[{task_id}] Detected stale in_progress state (likely crash), allowing resume")
            # Continue with resume logic...

Option B: Timeout-based detection

Simpler but less accurate - check last_update timestamp in import_state:

def resume_import(...):
    # ... after getting status_result ...

    # Check for stale in_progress
    for row in status_result:
        if row.get("status") == "in_progress":
            last_update = row.get("time")
            age_seconds = (time.time() * 1e9 - last_update) / 1e9
            if age_seconds > 60:  # Stale threshold
                influxdb3_local.warn(
                    f"[{task_id}] Import has stale in_progress state "
                    f"(last update {age_seconds:.0f}s ago), treating as crashed"
                )
                # Allow resume...

Option C: Add CRASHED state to ImportPauseState

Extend the enum and detection logic:

class ImportPauseState(Enum):
    NOT_FOUND = "not_found"
    CANCELLED = "cancelled"
    PAUSED = "paused"
    RUNNING = "running"
    COMPLETED = "completed"  # New
    CRASHED = "crashed"      # New

Affected Code

• File: import.py
• Functions:
  • get_import_pause_state() - needs to detect stale state
  • resume_import() - needs to handle crashed imports
  • import_table() - needs to write heartbeats (if Option A)
• Lines: 2250-2288 (get_import_pause_state), 2320-2475 (resume_import)

Workaround

Currently, the only workaround is to manually update the database:

-- Set the import to paused state so it can be resumed
INSERT INTO import_pause_state (import_id, paused, canceled, time)
VALUES ('<import_id>', true, false, now())

Then resume will work (though it may restart from beginning if paused_at_time was not captured).