Skip to content

fix: treat missing tarball objects as permanent and defer superseded deletes until commit #356

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hieptl merged 1 commit into from
Aug 21, 2026
Merged

fix: treat missing tarball objects as permanent and defer superseded deletes until commit #356

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions openhands/automation/dispatcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,8 +77,9 @@ async def _download_internal_tarball(
"""Download a tarball from storage using the TarballUpload record.

Raises:
TarballNotFoundError: If the tarball upload record doesn't exist.
This is a permanent error that should disable the automation.
TarballNotFoundError: If the tarball upload record doesn't exist, or if
the record exists but its storage object is missing. Both are
permanent errors that should disable the automation.
ValueError: If no database session is provided.
"""
if session is None:
Expand All @@ -94,10 +95,19 @@ async def _download_internal_tarball(
"The tarball may have been deleted."
)

from openhands.automation.storage import get_file_store
from openhands.automation.storage import ObjectNotFoundError, get_file_store

store = get_file_store()
return store.read(upload.storage_path)
try:
return store.read(upload.storage_path)
except ObjectNotFoundError as e:
# Only confirmed absence is permanent; transient storage errors raise
# plain FileNotFoundError and keep retrying on the next schedule tick.
raise TarballNotFoundError(
f"Internal tarball object missing from storage at "
f"{upload.storage_path!r} for upload {upload_id}. "
"Recreate the automation to restore it."
) from e


async def _poll_pending_runs(
Expand Down
93 changes: 69 additions & 24 deletions openhands/automation/git_sync/loop.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@
UploadStatus,
)
from openhands.automation.schemas import Trigger, validate_command_string
from openhands.automation.storage import get_file_store
from openhands.automation.storage import ObjectNotFoundError, get_file_store
from openhands.automation.utils import utcnow
from openhands.automation.utils.periodic_loop import run_periodic_loop
from openhands.automation.utils.service_metadata import (
Expand Down Expand Up @@ -357,13 +357,20 @@ async def _stored_tarball_matches(
return False


async def _delete_superseded_upload(session: AsyncSession, tarball_path: str) -> None:
"""Remove the upload an automation just stopped pointing at.
async def _delete_superseded_upload(
session: AsyncSession, tarball_path: str, pending_storage_deletes: list[str]
) -> None:
"""Mark the upload an automation just stopped pointing at for removal.

Without this, every git-side edit left the previous upload row and file
referenced by nothing, so routine PR edits accumulated a tarball copy per
merge with no reaper to collect them. Mirrors the cleanup
`regenerate_preset_prompt_tarball` does for prompt edits.

Soft-deletes the record in the current transaction; the storage object is
only queued on `pending_storage_deletes` and removed after the cycle's
commit. Deleting before the commit destroyed the object irreversibly while
a rollback revived the record, stranding it pointing at a missing object.
"""
upload_id = parse_internal_upload_id(tarball_path)
if upload_id is None:
Expand All @@ -375,21 +382,22 @@ async def _delete_superseded_upload(session: AsyncSession, tarball_path: str) ->
if upload is None or upload.deleted_at is not None:
return

file_removed = False
try:
await asyncio.to_thread(get_file_store().delete, upload.storage_path)
file_removed = True
except FileNotFoundError:
file_removed = True
except Exception:
logger.exception(
"Failed to delete superseded tarball at %s", upload.storage_path
)
# Soft-delete only once the file is confirmed gone: a failed delete leaves
# the record live, so the file stays discoverable for a retry rather than
# becoming a hidden orphan.
if file_removed:
upload.deleted_at = utcnow()
upload.deleted_at = utcnow()
pending_storage_deletes.append(upload.storage_path)


async def _delete_pending_storage_objects(storage_paths: list[str]) -> None:
"""Best-effort removal of superseded tarball objects after the commit."""
if not storage_paths:
return
file_store = get_file_store()
for path in storage_paths:
try:
await asyncio.to_thread(file_store.delete, path)
except ObjectNotFoundError:
pass # already gone
except Exception:
logger.exception("Failed to delete superseded tarball at %s", path)


async def _resolve_tarball_path(
Expand All @@ -398,6 +406,7 @@ async def _resolve_tarball_path(
deserialized: DeserializedAutomation,
slug: str,
existing: Automation | None,
pending_storage_deletes: list[str],
) -> str:
if deserialized.tarball_bytes is not None:
if existing is not None and await _stored_tarball_matches(
Expand All @@ -417,7 +426,9 @@ async def _resolve_tarball_path(
slug,
)
if existing is not None:
await _delete_superseded_upload(session, existing.tarball_path)
await _delete_superseded_upload(
session, existing.tarball_path, pending_storage_deletes
)
return new_path

tarball_source = fields.get("tarball_source") or {}
Expand All @@ -439,6 +450,7 @@ async def _validate_and_resolve_fields(
fields: dict,
deserialized: DeserializedAutomation,
slug: str,
pending_storage_deletes: list[str],
existing: Automation | None = None,
) -> dict:
"""Validate automation.yaml fields and resolve a tarball_path, mirroring
Expand All @@ -459,7 +471,7 @@ async def _validate_and_resolve_fields(
)
timeout = validate_automation_timeout(fields.get("timeout"))
tarball_path = await _resolve_tarball_path(
session, fields, deserialized, slug, existing
session, fields, deserialized, slug, existing, pending_storage_deletes
)

return {
Expand All @@ -486,9 +498,10 @@ async def _create_automation_from_git(
deserialized: DeserializedAutomation,
dir_files: dict[str, bytes],
head: str,
pending_storage_deletes: list[str],
) -> None:
values = await _validate_and_resolve_fields(
session, deserialized.fields, deserialized, slug
session, deserialized.fields, deserialized, slug, pending_storage_deletes
)
local_user = _get_local_user()
automation = Automation(
Expand Down Expand Up @@ -516,6 +529,7 @@ async def _update_automation_from_git(
deserialized: DeserializedAutomation,
dir_files: dict[str, bytes],
head: str,
pending_storage_deletes: list[str],
) -> None:
new_hash = compute_content_hash(dir_files)
if new_hash == state.content_hash:
Expand All @@ -531,7 +545,12 @@ async def _update_automation_from_git(
)

values = await _validate_and_resolve_fields(
session, deserialized.fields, deserialized, state.slug, existing=automation
session,
deserialized.fields,
deserialized,
state.slug,
pending_storage_deletes,
existing=automation,
)
for column, value in values.items():
setattr(automation, column, value)
Expand Down Expand Up @@ -612,6 +631,7 @@ async def _import_from_git(
timeout: float,
encryption_key: str,
result: SyncCycleResult,
pending_storage_deletes: list[str],
) -> None:
all_dirs = _list_slug_directories(sync_root)

Expand All @@ -635,6 +655,9 @@ async def _import_from_git(
continue

dir_files = await asyncio.to_thread(_read_directory_files, directory)
# A rolled-back savepoint revives the soft-deleted upload rows queued
# inside it, so their storage paths must not stay pending for deletion.
pending_snapshot = len(pending_storage_deletes)
try:
# One SAVEPOINT per directory: a failure rolls back just its writes
# (a half-populated Automation, a flushed TarballUpload) and leaves
Expand All @@ -648,18 +671,30 @@ async def _import_from_git(

if state is None:
await _create_automation_from_git(
session, slug, deserialized, dir_files, head
session,
slug,
deserialized,
dir_files,
head,
pending_storage_deletes,
)
result.imported += 1
else:
await _update_automation_from_git(
session, state, deserialized, dir_files, head
session,
state,
deserialized,
dir_files,
head,
pending_storage_deletes,
)
except (ValidationError, ValueError) as e:
del pending_storage_deletes[pending_snapshot:]
logger.warning(
"Skipping invalid automation directory %r from git: %s", slug, e
)
except Exception:
del pending_storage_deletes[pending_snapshot:]
# Deliberately broad: anything escaping aborts the whole cycle --
# no export, no push -- every cycle until someone fixes that one
# directory by hand. Traceback logged, since reaching here means an
Expand Down Expand Up @@ -945,6 +980,11 @@ async def _run_sync_cycle_locked(

result = SyncCycleResult(head=head)

# Storage paths of superseded uploads soft-deleted during the import; their
# objects are only removed after the commit below succeeds, so a rollback
# can never revive a record whose object is already gone.
pending_storage_deletes: list[str] = []

# Committed *before* the push, not held open across it: SQLite holds its
# single-writer lock for the whole transaction, and a push can take up to
# git_sync_git_timeout_seconds.
Expand All @@ -970,6 +1010,7 @@ async def _run_sync_cycle_locked(
timeout,
encryption_key,
result,
pending_storage_deletes,
)

# After the import, so automations that arrived from git this cycle
Expand All @@ -981,6 +1022,10 @@ async def _run_sync_cycle_locked(
await _export_dirty_automations(session, sync_root, encryption_key, result)
await session.commit()

# Only after the commit: the soft-deletes are durable, so removing the
# objects can no longer strand a live record. A commit failure skips this.
await _delete_pending_storage_objects(pending_storage_deletes)

# Unconditional, not gated on `exported`: commit_and_push also retries a
# previous cycle's unpushed commit and pushes to a newly-repointed remote.
# Gating made those recovery paths unreachable whenever there was nothing
Expand Down
74 changes: 43 additions & 31 deletions openhands/automation/preset_router.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,15 @@
from pathlib import Path
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi import (
APIRouter,
BackgroundTasks,
Depends,
HTTPException,
Request,
Response,
status,
)
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
Expand All @@ -34,7 +42,7 @@
TemplateProvenance,
Trigger,
)
from openhands.automation.storage import FileStore, get_file_store
from openhands.automation.storage import FileStore, ObjectNotFoundError, get_file_store
from openhands.automation.telemetry import (
capture_automation_event,
get_request_telemetry_context,
Expand Down Expand Up @@ -298,10 +306,23 @@ def _replace_prompt_in_tarball(tarball_bytes: bytes, new_prompt: str) -> bytes |
return out_buffer.read()


def _delete_storage_object_best_effort(
file_store: FileStore, storage_path: str
) -> None:
"""Best-effort post-commit removal of a superseded tarball object."""
try:
file_store.delete(storage_path)
except ObjectNotFoundError:
pass # already gone -- nothing to clean up
except Exception:
logger.exception("Failed to delete superseded tarball at %s", storage_path)


async def regenerate_preset_prompt_tarball(
automation: Automation,
new_prompt: str,
session: AsyncSession,
background_tasks: BackgroundTasks,
) -> str | None:
"""Rebuild a preset automation's tarball with an updated prompt.

Expand All @@ -313,6 +334,8 @@ async def regenerate_preset_prompt_tarball(
Reads the automation's current internal-upload tarball, swaps in ``new_prompt``
(leaving all other files untouched), uploads the result as a new internal upload,
and returns its ``oh-internal://`` URL for the caller to store on ``tarball_path``.
The superseded upload is soft-deleted in the current transaction; its storage
object is removed via ``background_tasks`` only after the transaction commits.

Returns ``None`` — leaving the tarball unchanged — when the automation is not a
regenerable preset: its ``tarball_path`` is an external URL, the referenced upload
Expand All @@ -333,7 +356,10 @@ async def regenerate_preset_prompt_tarball(

try:
current_tarball = file_store.read(source_upload.storage_path)
except FileNotFoundError:
except ObjectNotFoundError:
# Only confirmed absence means "not regenerable"; a transient storage
# error must propagate (rolling back the edit) rather than silently
# leaving the old prompt baked into the tarball.
return None

new_tarball = _replace_prompt_in_tarball(current_tarball, new_prompt)
Expand Down Expand Up @@ -376,46 +402,32 @@ async def regenerate_preset_prompt_tarball(
detail=f"Failed to upload regenerated tarball: {e!s}",
)

# The old tarball is now superseded. Remove its file and soft-delete the
# upload record so repeated prompt edits don't accumulate orphaned storage.
# Only soft-delete once the file is confirmed gone: if the delete fails the
# record stays live so the still-present file remains discoverable for a
# later retry/cleanup instead of becoming a hidden orphan (file on disk,
# record marked deleted).
old_object_delete_succeeded = False
old_object_already_missing = False
try:
file_store.delete(source_upload.storage_path)
old_object_delete_succeeded = True
except FileNotFoundError:
old_object_already_missing = True
except Exception as e:
logger.exception(
"Failed to delete superseded tarball at %s: %s",
source_upload.storage_path,
e,
)
file_removed = old_object_delete_succeeded or old_object_already_missing
if file_removed:
source_upload.deleted_at = utcnow()
# The old tarball is now superseded. Soft-delete its record inside this
# transaction, but remove its storage object only after the commit, via a
# background task (which runs only for a successful response, after the
# function-scoped session has committed -- see update_automation). Deleting
# before the commit destroyed the object irreversibly while a rollback
# reverted this soft-delete and the tarball_path update, stranding a live
# record pointing at a missing object. Worst case now -- a crash between
# commit and task -- leaks an orphaned object whose record is already
# soft-deleted, which is the recoverable direction.
source_upload.deleted_at = utcnow()
background_tasks.add_task(
_delete_storage_object_best_effort, file_store, source_upload.storage_path
)

logger.info(
"Regenerated preset tarball: automation_id=%s, old_upload_id=%s, "
"new_upload_id=%s, old_object_delete_succeeded=%s, "
"old_object_already_missing=%s",
"new_upload_id=%s",
automation.id,
source_upload.id,
new_upload_id,
old_object_delete_succeeded,
old_object_already_missing,
extra={
"automation_id": str(automation.id),
"old_upload_id": str(source_upload.id),
"old_storage_path": source_upload.storage_path,
"new_upload_id": str(new_upload_id),
"new_storage_path": storage_path,
"old_object_delete_succeeded": old_object_delete_succeeded,
"old_object_already_missing": old_object_already_missing,
},
)

Expand Down
Loading
Loading