fix: harden deployment list sync and SQLite timestamp storage

- Filter list/count and sync by deployment_type in SQL so type-scoped
  sync can prune provider-unknown rows without a Python type skip.
- Keep list sync alive on provider round failures; backfill remaining
  page slots from local rows after sync (may be stale; tune batch size /
  max rounds to reduce padding).
- Bind attachment cleanup/recount to provider-confirmed IDs only (no
  needless snapshot copy before backfill).
- Soften DELETE rowcount handling with UnknownDeleteCount when the
  driver does not report an int.
- Document that deployment_list_sync_batch_size should stay within the
  provider's ID-filtered list limit.
- Stop writing deployment-family timestamps with server_default=func.now();
  use Column default/onupdate=utc_now (Field stays default=None) so SQLite
  stores microsecond DateTime strings matching later query parameters.
- Add SQLite migration a8f3c2d1e4b5: rewrite existing second-level timestamp
  strings through DateTime binds, and drop CURRENT_TIMESTAMP server defaults
  on all dialects.
- Cover type filter, keyset after rewrite, ORM microsecond writes, sync
  backfill/failure paths, and the server_default drop in unit tests; extend
  integration coverage for the sync path.

Author: HzaRashid

src/backend/base/langflow/api/v1/mappers/deployments/helpers.py

@@ -892,7 +892,9 @@ async def list_deployments_synced(
 
     Each round fetches a candidate window so one provider lookup reconciles
     stale rows beyond the page. Refill rounds seek after the last processed row
-    to avoid corruption to the offset under concurrent deletes.
+    to avoid offset drift under concurrent deletes. After provider sync finishes,
+    any remaining page slots are backfilled from local rows without another
+    provider check (they may be stale).
 
     ``allowed_ids`` is the DB-layer authorization prefilter, threaded into both
     the page query and the total count so a registered authorization plugin can
@@ -915,46 +917,57 @@ async def list_deployments_synced(
     for _ in range(max_sync_rounds):
         if len(accepted) >= size:
             break
-        batch = await list_deployments_page(
-            db,
-            user_id=user_id,
-            deployment_provider_account_id=provider_id,
-            offset=initial_offset if not cursor_created_at else None,
-            limit=sync_batch_size,
-            flow_version_ids=flow_version_ids,
-            project_id=project_id,
-            allowed_ids=allowed_ids,
-            cursor_created_at=cursor_created_at,
-            cursor_exclude_id=cursor_exclude_id,
-        )
-        if not batch:
-            break
+        try:
+            batch = await list_deployments_page(
+                db,
+                user_id=user_id,
+                deployment_provider_account_id=provider_id,
+                offset=initial_offset if cursor_created_at is None else None,
+                limit=sync_batch_size,
+                flow_version_ids=flow_version_ids,
+                project_id=project_id,
+                deployment_type=deployment_type,
+                allowed_ids=allowed_ids,
+                cursor_created_at=cursor_created_at,
+                cursor_exclude_id=cursor_exclude_id,
+            )
+            if not batch:
+                break
 
-        known, provider_view = await fetch_provider_resource_keys(
-            deployment_adapter=deployment_adapter,
-            user_id=user_id,
-            provider_id=provider_id,
-            db=db,
-            resource_keys=[row.resource_key for row, _, _ in batch],
-            deployment_type=deployment_type,
-        )
-        provider_bindings.extend(deployment_mapper.extract_snapshot_bindings(provider_view))
-        provider_data_by_resource_key.update(deployment_mapper.extract_list_item_provider_data(provider_view))
-        provider_metadata_by_resource_key.update(deployment_mapper.extract_metadata_for_list(provider_view))
-
-        last_row = batch[-1][0]
-        cursor_created_at = last_row.created_at
-        cursor_exclude_id = last_row.id
-        for row, attached_count, matched_flow_versions in batch:
-            if row.resource_key not in known:
-                # Provider `known` is type-filtered; skip other local types instead of deleting as stale.
-                if deployment_type is not None and row.deployment_type != deployment_type:
+            known, provider_view = await fetch_provider_resource_keys(
+                deployment_adapter=deployment_adapter,
+                user_id=user_id,
+                provider_id=provider_id,
+                db=db,
+                resource_keys=[row.resource_key for row, _, _ in batch],
+                deployment_type=deployment_type,
+            )
+            provider_bindings.extend(deployment_mapper.extract_snapshot_bindings(provider_view))
+            provider_data_by_resource_key.update(deployment_mapper.extract_list_item_provider_data(provider_view))
+            provider_metadata_by_resource_key.update(deployment_mapper.extract_metadata_for_list(provider_view))
+
+            last_row = batch[-1][0]
+            cursor_created_at = last_row.created_at
+            cursor_exclude_id = last_row.id
+            for row, attached_count, matched_flow_versions in batch:
+                if row.resource_key not in known:
+                    # Page rows are already type-filtered in SQL when deployment_type
+                    # is set, so provider-unknown rows are safe to prune as stale.
+                    stale_deployment_owner_pairs.append(DeploymentOwnerPair(owner_id=row.user_id, deployment_id=row.id))
                     continue
-                stale_deployment_owner_pairs.append(DeploymentOwnerPair(owner_id=row.user_id, deployment_id=row.id))
-                continue
-            if len(accepted) < size:
-                accepted.append((row, attached_count, matched_flow_versions))
-                accepted_deployment_ids.append(row.id)
+                if len(accepted) < size:
+                    accepted.append((row, attached_count, matched_flow_versions))
+                    accepted_deployment_ids.append(row.id)
+        except Exception:  # noqa: BLE001
+            # Keep the list path alive: return whatever earlier rounds accepted
+            # and still prune confirmed stales once below.
+            logger.warning(
+                "Deployment list sync round failed for provider %s; continuing with %d accepted row(s)",
+                provider_id,
+                len(accepted),
+                exc_info=True,
+            )
+            break
 
     if stale_deployment_owner_pairs:
         logger.warning(
@@ -964,7 +977,7 @@ async def list_deployments_synced(
         )
         await delete_deployments_by_owner_and_ids(db, deployment_owner_pairs=stale_deployment_owner_pairs)
 
-    # Phase 2: metadata and binding-level sync.
+    # Phase 2: metadata and binding-level sync (provider-confirmed rows only).
     if accepted:
         metadata_updates: list[DeploymentMetadataUpdate] = []
         for row, _attached_count, _matched in accepted:
@@ -982,7 +995,8 @@ async def list_deployments_synced(
 
     # Remove stale local attachments based on provider bindings, then recount.
     # Best-effort - provider or DB failures should not block the list response.
-    if accepted:
+    # Only provider-confirmed IDs: backfill rows below are not binding-synced.
+    if accepted_deployment_ids:
         try:
             async with db.begin_nested():
                 await delete_unbound_attachments(
@@ -1005,12 +1019,40 @@ async def list_deployments_synced(
                 exc_info=True,
             )
 
+    # Pad the response page from local rows after all provider sync is done.
+    # These rows may be stale and are not metadata/binding-synced above.
+    remaining = size - len(accepted)
+    if remaining > 0:
+        try:
+            backfill_batch = await list_deployments_page(
+                db,
+                user_id=user_id,
+                deployment_provider_account_id=provider_id,
+                offset=initial_offset if cursor_created_at is None else None,
+                limit=remaining,
+                flow_version_ids=flow_version_ids,
+                project_id=project_id,
+                deployment_type=deployment_type,
+                allowed_ids=allowed_ids,
+                cursor_created_at=cursor_created_at,
+                cursor_exclude_id=cursor_exclude_id,
+            )
+            accepted.extend(backfill_batch)
+        except Exception:  # noqa: BLE001
+            logger.warning(
+                "Deployment list backfill failed for provider %s; returning %d accepted row(s)",
+                provider_id,
+                len(accepted),
+                exc_info=True,
+            )
+
     total = await count_deployments_by_provider(
         db,
         user_id=user_id,
         deployment_provider_account_id=provider_id,
         flow_version_ids=flow_version_ids,
         project_id=project_id,
+        deployment_type=deployment_type,
         allowed_ids=allowed_ids,
     )
     return accepted, total, provider_data_by_resource_key

src/backend/base/langflow/services/database/models/deployment/crud.py

@@ -5,7 +5,7 @@
 from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
 
 from lfx.log.logger import logger
-from sqlalchemy import Select, column, or_, tuple_, values
+from sqlalchemy import Select, and_, column, or_, tuple_, values
 from sqlalchemy.exc import IntegrityError
 from sqlmodel import col, delete, func, select, update
 
@@ -36,6 +36,21 @@ class DeploymentOwnerPair(NamedTuple):
     deployment_id: UUID
 
 
+class UnknownDeleteCount:
+    """Sentinel returned when DELETE rowcount is not a usable integer."""
+
+    __slots__ = ()
+
+    def __repr__(self) -> str:
+        return "UnknownDeleteCount()"
+
+    def __bool__(self) -> bool:
+        return False
+
+
+UNKNOWN_DELETE_COUNT = UnknownDeleteCount()
+
+
 @dataclass(frozen=True, slots=True)
 class DeploymentMetadataUpdate:
     langflow_db_row: Deployment
@@ -365,6 +380,7 @@ async def list_deployments_page(
     offset: int | None = None,
     flow_version_ids: list[UUID] | None = None,
     project_id: UUID | None = None,
+    deployment_type: DeploymentType | None = None,
     allowed_ids: list[UUID] | None = None,
     cursor_created_at: datetime | None = None,
     cursor_exclude_id: UUID | None = None,
@@ -381,9 +397,13 @@ async def list_deployments_page(
     authorization plugin reports the caller may read) — see
     ``langflow.services.authorization.restrict_to_owned_or_visible``.
 
+    ``deployment_type`` optionally restricts the page to one local deployment
+    type. When set, callers that also type-filter the provider can treat
+    provider-unknown rows as stale without a Python type skip.
+
     Use ``offset`` for the first page. Use ``cursor_created_at`` and
     ``cursor_exclude_id`` together for follow-up keyset reads; cursor reads must
-    not also pass ``offset``.
+    not also pass ``offset``. Exactly one of those modes is required.
     """
     if offset is not None and offset < 0:
         msg = "offset must be greater than or equal to 0"
@@ -394,9 +414,12 @@ async def list_deployments_page(
     if (cursor_created_at is None) != (cursor_exclude_id is None):
         msg = "cursor_created_at and cursor_exclude_id must be provided together"
         raise ValueError(msg)
-    if cursor_created_at and offset is not None:
+    if cursor_created_at is not None and offset is not None:
         msg = "offset cannot be used with cursor_created_at and cursor_exclude_id"
         raise ValueError(msg)
+    if offset is None and cursor_created_at is None:
+        msg = "either offset or cursor_created_at/cursor_exclude_id is required"
+        raise ValueError(msg)
     attachment_counts_subquery = (
         select(
             col(FlowVersionDeploymentAttachment.deployment_id).label("deployment_id"),
@@ -423,13 +446,17 @@ async def list_deployments_page(
     stmt = _scope_to_owner_or_allowed(stmt, user_id=user_id, allowed_ids=allowed_ids)
     if project_id is not None:
         stmt = stmt.where(Deployment.project_id == project_id)
-    if cursor_created_at:
+    if deployment_type is not None:
+        stmt = stmt.where(Deployment.deployment_type == deployment_type)
+    if cursor_created_at is not None:
+        # DESC keyset: seek strictly after (created_at, id) cursor.
         stmt = stmt.where(
             or_(
-                # older deployments
                 col(Deployment.created_at) < cursor_created_at,
-                # Same age but different deployments
-                (col(Deployment.created_at) == cursor_created_at) & (col(Deployment.id) < cursor_exclude_id),
+                and_(
+                    col(Deployment.created_at) == cursor_created_at,
+                    col(Deployment.id) < cursor_exclude_id,
+                ),
             )
         )
     if flow_version_ids:
@@ -571,20 +598,24 @@ async def count_deployments_by_provider(
     deployment_provider_account_id: UUID,
     flow_version_ids: list[UUID] | None = None,
     project_id: UUID | None = None,
+    deployment_type: DeploymentType | None = None,
     allowed_ids: list[UUID] | None = None,
 ) -> int:
     """Count deployments for a provider account.
 
     ``allowed_ids`` mirrors ``list_deployments_page``: ``None`` counts owner rows
     only (OSS default); a list counts the owner ⊕ ``allowed_ids`` union so the
     pagination total reflects the same authorization prefilter as the page.
+    ``deployment_type``, when set, counts only rows of that local type.
     """
     stmt = select(func.count(Deployment.id)).where(
         Deployment.deployment_provider_account_id == deployment_provider_account_id,
     )
     stmt = _scope_to_owner_or_allowed(stmt, user_id=user_id, allowed_ids=allowed_ids)
     if project_id is not None:
         stmt = stmt.where(Deployment.project_id == project_id)
+    if deployment_type is not None:
+        stmt = stmt.where(Deployment.deployment_type == deployment_type)
     if flow_version_ids:
         matched_deployments_subquery = (
             select(FlowVersionDeploymentAttachment.deployment_id)
@@ -713,8 +744,13 @@ async def delete_deployments_by_owner_and_ids(
     db: AsyncSession,
     *,
     deployment_owner_pairs: list[DeploymentOwnerPair],
-) -> int:
-    """Delete owner-scoped deployment rows; return the number of deployment rows deleted."""
+) -> int | UnknownDeleteCount:
+    """Delete owner-scoped deployment rows.
+
+    Returns:
+        The deleted row count when the driver reports an ``int`` rowcount;
+        otherwise ``UNKNOWN_DELETE_COUNT``.
+    """
     if not deployment_owner_pairs:
         return 0
 
@@ -735,14 +771,15 @@ async def delete_deployments_by_owner_and_ids(
             ).in_(deployment_owner_pairs),
         )
     )
-    if result.rowcount is None:
-        msg = (
-            "DELETE rowcount was None for deployments=%s -- "
-            "database driver may not support rowcount for DELETE statements"
-        )
-        await logger.aerror(
-            msg,
-            deployment_owner_pairs,
-        )
-        raise RuntimeError(msg % deployment_owner_pairs)
-    return result.rowcount
+    # Prefer isinstance over ``is not None``: some drivers may return non-int
+    # sentinels, and inventing a count from those would be misleading.
+    if isinstance(result.rowcount, int):
+        return result.rowcount
+
+    await logger.aerror(
+        "DELETE rowcount was not an int for deployments=%s (got %r) -- "
+        "database driver may not support rowcount for DELETE statements",
+        deployment_owner_pairs,
+        result.rowcount,
+    )
+    return UNKNOWN_DELETE_COUNT