Refactor calendar outbox delivery paths
This commit is contained in:
@@ -858,15 +858,18 @@ def _fail_operation(
|
||||
session.flush()
|
||||
|
||||
|
||||
def execute_calendar_outbox_operation(
|
||||
def _lock_leased_outbox_operation(
|
||||
session: Session,
|
||||
*,
|
||||
operation_id: str,
|
||||
lease_token: str,
|
||||
client_factory: Callable[[Session, CalendarSyncSource], object] | None = None,
|
||||
) -> CalendarOutboxOperation:
|
||||
) -> tuple[CalendarOutboxOperation, CalendarSyncSource | None]:
|
||||
operation = session.get(CalendarOutboxOperation, operation_id)
|
||||
if operation is None or operation.status != "in_progress" or operation.lease_token != lease_token:
|
||||
if (
|
||||
operation is None
|
||||
or operation.status != "in_progress"
|
||||
or operation.lease_token != lease_token
|
||||
):
|
||||
raise ValueError("Calendar outbox lease is no longer valid")
|
||||
source = (
|
||||
session.query(CalendarSyncSource)
|
||||
@@ -882,151 +885,282 @@ def execute_calendar_outbox_operation(
|
||||
)
|
||||
if operation.status != "in_progress" or operation.lease_token != lease_token:
|
||||
raise ValueError("Calendar outbox lease is no longer valid")
|
||||
operation_metadata = operation.metadata_ or {}
|
||||
unavailable_reason = _operation_source_unavailable_reason(operation, source)
|
||||
if unavailable_reason:
|
||||
if source is not None and source.tenant_id == operation.tenant_id:
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error=unavailable_reason,
|
||||
terminal_status="cancelled",
|
||||
)
|
||||
else:
|
||||
operation.status = "cancelled"
|
||||
operation.completed_at = utcnow()
|
||||
operation.last_error = unavailable_reason
|
||||
operation.lease_token = None
|
||||
operation.lease_expires_at = None
|
||||
session.flush()
|
||||
return operation
|
||||
return operation, source
|
||||
|
||||
overwrite = bool(operation_metadata.get("overwrite")) if isinstance(operation_metadata, dict) else False
|
||||
client: object | None = None
|
||||
try:
|
||||
if client_factory is None:
|
||||
from govoplan_calendar.backend.service import caldav_client_for_source
|
||||
|
||||
client = caldav_client_for_source(session, source)
|
||||
else:
|
||||
client = client_factory(session, source)
|
||||
if operation.operation_kind == "put":
|
||||
result = client.put_object(
|
||||
operation.resource_href,
|
||||
operation.payload_ics or "",
|
||||
etag=operation.expected_etag,
|
||||
create=not bool(operation.expected_etag),
|
||||
overwrite=overwrite,
|
||||
)
|
||||
remote_etag = result.etag
|
||||
reconciled = False
|
||||
if not remote_etag:
|
||||
try:
|
||||
matched, remote_etag = _remote_matches_operation(client, operation)
|
||||
except Exception:
|
||||
matched, remote_etag = False, None
|
||||
if not matched or not remote_etag:
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error="CalDAV PUT succeeded without an ETag and could not be reconciled safely",
|
||||
)
|
||||
return operation
|
||||
reconciled = True
|
||||
_complete_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_etag=remote_etag,
|
||||
reconciled=reconciled,
|
||||
)
|
||||
return operation
|
||||
|
||||
if not operation.expected_etag and not overwrite:
|
||||
matched, remote_etag = _remote_matches_operation(client, operation)
|
||||
if matched:
|
||||
_complete_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_etag=remote_etag,
|
||||
reconciled=True,
|
||||
)
|
||||
return operation
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error=(
|
||||
"CalDAV DELETE has no known ETag and the remote resource exists; "
|
||||
"refusing to delete an unknown remote version"
|
||||
),
|
||||
terminal_status="conflict",
|
||||
)
|
||||
return operation
|
||||
result = client.delete_object(
|
||||
operation.resource_href,
|
||||
etag=operation.expected_etag,
|
||||
overwrite=overwrite,
|
||||
def _cancel_undeliverable_operation(
|
||||
session: Session,
|
||||
*,
|
||||
operation: CalendarOutboxOperation,
|
||||
source: CalendarSyncSource | None,
|
||||
reason: str,
|
||||
) -> None:
|
||||
if source is not None and source.tenant_id == operation.tenant_id:
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error=reason,
|
||||
terminal_status="cancelled",
|
||||
)
|
||||
return
|
||||
operation.status = "cancelled"
|
||||
operation.completed_at = utcnow()
|
||||
operation.last_error = reason
|
||||
operation.lease_token = None
|
||||
operation.lease_expires_at = None
|
||||
session.flush()
|
||||
|
||||
|
||||
def _calendar_outbox_client(
|
||||
session: Session,
|
||||
*,
|
||||
source: CalendarSyncSource,
|
||||
client_factory: Callable[[Session, CalendarSyncSource], object] | None,
|
||||
) -> object:
|
||||
if client_factory is not None:
|
||||
return client_factory(session, source)
|
||||
from govoplan_calendar.backend.service import caldav_client_for_source
|
||||
|
||||
return caldav_client_for_source(session, source)
|
||||
|
||||
|
||||
def _try_remote_match(
|
||||
client: object,
|
||||
operation: CalendarOutboxOperation,
|
||||
) -> tuple[bool, str | None] | None:
|
||||
try:
|
||||
return _remote_matches_operation(client, operation)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _complete_if_remote_match_is_usable(
|
||||
session: Session,
|
||||
*,
|
||||
operation: CalendarOutboxOperation,
|
||||
source: CalendarSyncSource,
|
||||
remote_match: tuple[bool, str | None] | None,
|
||||
) -> bool:
|
||||
if remote_match is None:
|
||||
return False
|
||||
matched, remote_etag = remote_match
|
||||
if not matched or (operation.operation_kind != "delete" and not remote_etag):
|
||||
return False
|
||||
_complete_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_etag=remote_etag,
|
||||
reconciled=True,
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def _execute_calendar_outbox_put(
|
||||
session: Session,
|
||||
*,
|
||||
operation: CalendarOutboxOperation,
|
||||
source: CalendarSyncSource,
|
||||
client: object,
|
||||
overwrite: bool,
|
||||
) -> None:
|
||||
result = client.put_object(
|
||||
operation.resource_href,
|
||||
operation.payload_ics or "",
|
||||
etag=operation.expected_etag,
|
||||
create=not bool(operation.expected_etag),
|
||||
overwrite=overwrite,
|
||||
)
|
||||
remote_etag = result.etag
|
||||
if remote_etag:
|
||||
_complete_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_etag=result.etag,
|
||||
reconciled=result.status == 404,
|
||||
remote_etag=remote_etag,
|
||||
reconciled=False,
|
||||
)
|
||||
return
|
||||
remote_match = _try_remote_match(client, operation)
|
||||
if _complete_if_remote_match_is_usable(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_match=remote_match,
|
||||
):
|
||||
return
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error="CalDAV PUT succeeded without an ETag and could not be reconciled safely",
|
||||
)
|
||||
|
||||
|
||||
def _execute_calendar_outbox_delete(
|
||||
session: Session,
|
||||
*,
|
||||
operation: CalendarOutboxOperation,
|
||||
source: CalendarSyncSource,
|
||||
client: object,
|
||||
overwrite: bool,
|
||||
) -> None:
|
||||
if not operation.expected_etag and not overwrite:
|
||||
remote_match = _remote_matches_operation(client, operation)
|
||||
if _complete_if_remote_match_is_usable(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_match=remote_match,
|
||||
):
|
||||
return
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error=(
|
||||
"CalDAV DELETE has no known ETag and the remote resource exists; "
|
||||
"refusing to delete an unknown remote version"
|
||||
),
|
||||
terminal_status="conflict",
|
||||
)
|
||||
return
|
||||
result = client.delete_object(
|
||||
operation.resource_href,
|
||||
etag=operation.expected_etag,
|
||||
overwrite=overwrite,
|
||||
)
|
||||
_complete_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_etag=result.etag,
|
||||
reconciled=result.status == 404,
|
||||
)
|
||||
|
||||
|
||||
def _handle_outbox_precondition_failure(
|
||||
session: Session,
|
||||
*,
|
||||
operation: CalendarOutboxOperation,
|
||||
source: CalendarSyncSource,
|
||||
client: object | None,
|
||||
error: CalDAVPreconditionFailed,
|
||||
) -> None:
|
||||
remote_match = _try_remote_match(client, operation) if client is not None else None
|
||||
if _complete_if_remote_match_is_usable(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_match=remote_match,
|
||||
):
|
||||
return
|
||||
if remote_match is None:
|
||||
_fail_operation(session, operation=operation, source=source, error=error)
|
||||
return
|
||||
matched, _remote_etag = remote_match
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error=(
|
||||
"Matching CalDAV resource has no usable ETag; retrying reconciliation"
|
||||
if matched
|
||||
else "CalDAV resource changed remotely; reconcile or sync before retrying"
|
||||
),
|
||||
terminal_status=None if matched else "conflict",
|
||||
)
|
||||
|
||||
|
||||
def _handle_outbox_delivery_failure(
|
||||
session: Session,
|
||||
*,
|
||||
operation: CalendarOutboxOperation,
|
||||
source: CalendarSyncSource,
|
||||
client: object | None,
|
||||
error: Exception,
|
||||
) -> None:
|
||||
remote_match = _try_remote_match(client, operation) if client is not None else None
|
||||
if _complete_if_remote_match_is_usable(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_match=remote_match,
|
||||
):
|
||||
return
|
||||
_fail_operation(session, operation=operation, source=source, error=error)
|
||||
|
||||
|
||||
def execute_calendar_outbox_operation(
|
||||
session: Session,
|
||||
*,
|
||||
operation_id: str,
|
||||
lease_token: str,
|
||||
client_factory: Callable[[Session, CalendarSyncSource], object] | None = None,
|
||||
) -> CalendarOutboxOperation:
|
||||
operation, source = _lock_leased_outbox_operation(
|
||||
session,
|
||||
operation_id=operation_id,
|
||||
lease_token=lease_token,
|
||||
)
|
||||
unavailable_reason = _operation_source_unavailable_reason(operation, source)
|
||||
if unavailable_reason:
|
||||
_cancel_undeliverable_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
reason=unavailable_reason,
|
||||
)
|
||||
return operation
|
||||
except CalDAVPreconditionFailed as exc:
|
||||
if client is None:
|
||||
_fail_operation(session, operation=operation, source=source, error=exc)
|
||||
return operation
|
||||
try:
|
||||
matched, remote_etag = _remote_matches_operation(client, operation)
|
||||
except Exception:
|
||||
_fail_operation(session, operation=operation, source=source, error=exc)
|
||||
else:
|
||||
if matched and (operation.operation_kind == "delete" or remote_etag):
|
||||
_complete_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_etag=remote_etag,
|
||||
reconciled=True,
|
||||
)
|
||||
else:
|
||||
_fail_operation(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
error=(
|
||||
"Matching CalDAV resource has no usable ETag; retrying reconciliation"
|
||||
if matched
|
||||
else "CalDAV resource changed remotely; reconcile or sync before retrying"
|
||||
),
|
||||
terminal_status=None if matched else "conflict",
|
||||
)
|
||||
return operation
|
||||
except Exception as exc:
|
||||
matched, remote_etag = False, None
|
||||
if client is not None:
|
||||
try:
|
||||
matched, remote_etag = _remote_matches_operation(client, operation)
|
||||
except Exception:
|
||||
matched, remote_etag = False, None
|
||||
if matched and (operation.operation_kind == "delete" or remote_etag):
|
||||
_complete_operation(
|
||||
|
||||
assert source is not None
|
||||
operation_metadata = operation.metadata_ or {}
|
||||
overwrite = (
|
||||
bool(operation_metadata.get("overwrite"))
|
||||
if isinstance(operation_metadata, dict)
|
||||
else False
|
||||
)
|
||||
client: object | None = None
|
||||
try:
|
||||
client = _calendar_outbox_client(
|
||||
session,
|
||||
source=source,
|
||||
client_factory=client_factory,
|
||||
)
|
||||
if operation.operation_kind == "put":
|
||||
_execute_calendar_outbox_put(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
remote_etag=remote_etag,
|
||||
reconciled=True,
|
||||
client=client,
|
||||
overwrite=overwrite,
|
||||
)
|
||||
else:
|
||||
_fail_operation(session, operation=operation, source=source, error=exc)
|
||||
return operation
|
||||
_execute_calendar_outbox_delete(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
client=client,
|
||||
overwrite=overwrite,
|
||||
)
|
||||
except CalDAVPreconditionFailed as exc:
|
||||
_handle_outbox_precondition_failure(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
client=client,
|
||||
error=exc,
|
||||
)
|
||||
except Exception as exc:
|
||||
_handle_outbox_delivery_failure(
|
||||
session,
|
||||
operation=operation,
|
||||
source=source,
|
||||
client=client,
|
||||
error=exc,
|
||||
)
|
||||
return operation
|
||||
|
||||
|
||||
def dispatch_calendar_outbox(
|
||||
|
||||
Reference in New Issue
Block a user