3596         updated_at_col = func.coalesce(Project.updated_at, _KEYSET_PAGINATION_EPOCH)
  3597         if updated_after is not None:
  3598             query = query.filter(updated_at_col >= updated_after)
  3599         if keyset_after is not None or limit is not None:
! 3600             if keyset_after is not None:
! 3601                 after_updated_at, after_name = keyset_after
  3602                 # Portable keyset predicate (works identically on SQLite/Postgres/MySQL)
  3603                 # — avoids row-value comparison syntax some dialects handle inconsistently.
! 3604                 query = query.filter(
  3605                     or_(
  3606                         updated_at_col > after_updated_at,
  3607                         and_(
  3608                             updated_at_col == after_updated_at,

  3612                 )
  3613             # A stable order is required for keyset pagination to mean anything — the
  3614             # very first page (keyset_after=None) still needs it, so the boundary row
  3615             # a caller computes its next cursor from is consistent across pages.
! 3616             query = query.order_by(updated_at_col, Project.name)
  3617         if limit is not None:
! 3618             query = query.limit(limit)
  3619         if for_update:
! 3620             query = query.with_for_update()
  3621 
  3622         project_records = query.all()
  3623         return mlrun.common.schemas.ProjectsOutput(
  3624             projects=self._format_projects(project_records, format_)

  193         Re-routes the /follower/projects commit-delete call to chief. Blocks until
  194         the full purge finishes, so the timeout needs to cover that, not the legacy
  195         chief-proxy default.
  196         """
! 197         return await self._proxy_request_to_chief(
  198             "DELETE",
  199             f"follower/projects/{name}",
  200             request,
  201             timeout=int(

  376 
  377         if response.status_code == requests.codes.accepted:
  378             # 202: accepted, still converging - poll for completion
  379             if not wait_for_completion:
! 380                 return True
  381             op_id = response.json()["status"]["opId"]
  382             self._wait_for_op(name, op_id, auth_info)
  383 
  384         return False

  476                 auth_info,
  477                 fatal_exceptions=(_OrcaActionFailedError,),
  478             )
  479         except _OrcaActionFailedError as exc:
! 480             raise mlrun.errors.MLRunRuntimeError(str(exc)) from exc
  481 
  482     def _verify_op_terminal(
  483         self,
  484         name: str,

  498             },
  499         )
  500         items = response.json().get("items", [])
  501         if not items:
! 502             raise mlrun.errors.MLRunRuntimeError(
  503                 f"No Orca sync-project action observed yet for project {name} (op_id={op_id})"
  504             )
  505         state = items[0].get("status", {}).get("state")
  506         if state == "failed":

  578                 correlation_id,
  579                 fatal_exceptions=(_OrcaActionFailedError,),
  580             )
  581         except _OrcaActionFailedError as exc:
! 582             raise mlrun.errors.MLRunRuntimeError(str(exc)) from exc
  583 
  584     def _verify_followers_sync_terminal(self, correlation_id: str) -> None:
  585         response = self._send_self_authenticated_request(
  586             mlrun.common.types.HTTPMethod.GET,

  594             },
  595         )
  596         items = response.json().get("items", [])
  597         if not items:
! 598             raise mlrun.errors.MLRunRuntimeError(
  599                 f"No Orca followers-sync action observed yet (correlation_id={correlation_id})"
  600             )
  601         state = items[0].get("status", {}).get("state")
  602         if state == "failed":

  616         wire = {"name": project.metadata.name}
  617         if project.spec.owner:
  618             wire["owner"] = project.spec.owner
  619         if project.spec.description:
! 620             wire["description"] = project.spec.description
  621         if project.metadata.labels:
! 622             wire["labels"] = project.metadata.labels
  623         if project.metadata.annotations:
! 624             wire["annotations"] = project.metadata.annotations
  625         return wire
  626 
  627     @staticmethod
  628     def _update_project_wire(

  635             "prevOpId": str(prev_op_id) if prev_op_id else None,
  636             "owner": project.spec.owner,
  637         }
  638         if project.spec.description:
! 639             wire["description"] = project.spec.description
  640         if project.metadata.labels:
! 641             wire["labels"] = project.metadata.labels
  642         if project.metadata.annotations:
! 643             wire["annotations"] = project.metadata.annotations
  644         return wire
  645 
  646     @staticmethod
  647     def _wire_to_project(body: dict) -> mlrun.common.schemas.Project:

  48         such window. Overridden by follower.Member. Safe to call on any deployment mode - the
  49         followers-sync-status endpoint calls this polymorphically without knowing which one is
  50         running.
  51         """
! 52         return True
  53 
  54     def ensure_project(
  55         self,
  56         db_session: sqlalchemy.orm.Session,

  28     Lets a worker ask the chief whether the chief's own followers-sync trigger (Follower
  29     Contract HLD checklist item 10) has reached a terminal state yet, instead of each replica
  30     independently triggering its own sweep against the leader.
  31     """
! 32     if (
  33         mlrun.mlconf.httpdb.clusterization.role
  34         != mlrun.common.schemas.ClusterizationRole.chief
  35     ):
! 36         chief_client = framework.utils.clients.chief.Client()
! 37         return {"ready": await chief_client.is_followers_sync_ready()}
  38 
! 39     return {
  40         "ready": framework.utils.singletons.project_member.get_project_member().is_followers_sync_ready()
  41     }

  44 def _to_follower_state(
  45     name: str, snapshot: mlrun.common.schemas.Project | None
  46 ) -> follower_schemas.FollowerProjectState:
  47     if snapshot is None:
! 48         return follower_schemas.FollowerProjectState(name=name)
  49     return follower_schemas.FollowerProjectState(
  50         name=snapshot.metadata.name,
  51         op_id=snapshot.status.op_id,
  52         sync_status=snapshot.status.state,

  60 async def prepare_create_project(
  61     name: str,
  62     body: follower_schemas.FollowerPrepareCreateRequest,
  63 ) -> follower_schemas.FollowerProjectState:
! 64     if body.metadata.name != name:
! 65         raise mlrun.errors.MLRunBadRequestError(
  66             "Path project name does not match metadata.name"
  67         )
! 68     project = mlrun.common.schemas.Project(
  69         metadata=body.metadata,
  70         spec=body.spec,
  71         status=mlrun.common.schemas.ProjectStatus(op_id=body.status.op_id),
  72     )
! 73     result = await mlrun.utils.run_in_threadpool(
  74         services.api.crud.Projects().prepare_create_project,
  75         project,
  76         body.status.op_id,
  77     )
! 78     return _to_follower_state(name, result)
  79 
  80 
  81 @router.post(
  82     "/follower/projects/{name}/commit-create",

  85 async def commit_create_project(
  86     name: str,
  87     body: follower_schemas.FollowerCommitCreateRequest,
  88 ) -> follower_schemas.FollowerProjectState:
! 89     result = await mlrun.utils.run_in_threadpool(
  90         services.api.crud.Projects().commit_create_project, name, body.status.op_id
  91     )
! 92     return _to_follower_state(name, result)
  93 
  94 
  95 @router.put(
  96     "/follower/projects/{name}",

   99 async def update_project(
  100     name: str,
  101     body: follower_schemas.FollowerUpdateRequest,
  102 ) -> follower_schemas.FollowerProjectState:
! 103     if body.metadata.name != name:
! 104         raise mlrun.errors.MLRunBadRequestError(
  105             "Path project name does not match metadata.name"
  106         )
! 107     project = mlrun.common.schemas.Project(
  108         metadata=body.metadata,
  109         spec=body.spec,
  110         status=mlrun.common.schemas.ProjectStatus(op_id=body.status.op_id),
  111     )
! 112     result = await mlrun.utils.run_in_threadpool(
  113         services.api.crud.Projects().update_project_follower,
  114         name,
  115         project,
  116         op_id=body.status.op_id,

  115         project,
  116         op_id=body.status.op_id,
  117         prev_op_id=body.prev_op_id,
  118     )
! 119     return _to_follower_state(name, result)
  120 
  121 
  122 @router.post(
  123     "/follower/projects/{name}/prepare-delete",

  126 async def prepare_delete_project(
  127     name: str,
  128     body: follower_schemas.FollowerPrepareDeleteRequest,
  129 ) -> follower_schemas.FollowerProjectState:
! 130     result = await mlrun.utils.run_in_threadpool(
  131         services.api.crud.Projects().prepare_delete_project,
  132         name,
  133         op_id=body.status.op_id,
  134         prev_op_id=body.prev_op_id,

  132         name,
  133         op_id=body.status.op_id,
  134         prev_op_id=body.prev_op_id,
  135     )
! 136     return _to_follower_state(name, result)
  137 
  138 
  139 @router.delete(
  140     "/follower/projects/{name}",

  146     request: fastapi.Request,
  147 ) -> follower_schemas.FollowerProjectState:
  148     # delete_project_resources deletes schedules, which run only on chief, so we
  149     # re-route to chief — same reason the legacy (non-follower) delete endpoint does.
! 150     if (
  151         mlrun.mlconf.httpdb.clusterization.role
  152         != mlrun.common.schemas.ClusterizationRole.chief
  153     ):
! 154         logger.info(
  155             "Requesting to commit-delete follower project, re-routing to chief",
  156             project=name,
  157         )
! 158         return (
  159             await framework.utils.clients.chief.Client().commit_delete_follower_project(
  160                 name=name, request=request
  161             )
  162         )

  160                 name=name, request=request
  161             )
  162         )
  163 
! 164     op_id = body.status.op_id
  165     # Validates internally (CAS/ordering/state) and, on success, purges the project's
  166     # resources and its row — a no-op if it's already gone (a previous call already
  167     # removed it) or a genuine retry with the same op_id (e.g. after a dropped
  168     # connection re-runs the purge rather than skipping it).
! 169     await mlrun.utils.run_in_threadpool(
  170         services.api.crud.Projects().commit_delete_project, name, op_id
  171     )
  172     # The row is gone at this point — sync_status reports the state it was in right
  173     # before removal, matching every other op's {name, op_id, sync_status} response.
! 174     return follower_schemas.FollowerProjectState(
  175         name=name,
  176         op_id=op_id,
  177         sync_status=mlrun.common.schemas.ProjectState.deleting,
  178     )

  189     db_session: sqlalchemy.orm.Session = fastapi.Depends(
  190         framework.api.deps.get_db_session
  191     ),
  192 ) -> follower_schemas.FollowerProjectStatesPage:
! 193     projects, next_cursor = await mlrun.utils.run_in_threadpool(
  194         services.api.crud.Projects().list_project_states,
  195         db_session,
  196         updated_after,
  197         cursor,

  196         updated_after,
  197         cursor,
  198         page_size,
  199     )
! 200     return follower_schemas.FollowerProjectStatesPage(
  201         projects=[
  202             _to_follower_state(project.metadata.name, project) for project in projects
  203         ],
  204         next_cursor=next_cursor,

  1157             # Not on the wire for this op (Orca's request has no updated_at field at
  1158             # all) — MLRun stamps its own local time for when it applied the change.
  1159             project.status.updated_at = datetime.datetime.now(datetime.UTC)
  1160             if existing:
! 1161                 self.store_project(session, name, project)
  1162             else:
  1163                 self.create_project(session, project)
  1164             return project
  1165         finally:

  1225                 incoming_op_id=op_id,
  1226                 prev_op_id=prev_op_id,
  1227             )
  1228             if outcome == follower_contract.ReplayOutcome.replay:
! 1229                 return existing
  1230             # No project payload on this call, and updated_at isn't on the wire at
  1231             # all — MLRun stamps its own local time for when it applied the change.
  1232             updated_at = datetime.datetime.now(datetime.UTC)
  1233             self.patch_project(

  1309                 incoming_op_id=op_id,
  1310                 prev_op_id=prev_op_id,
  1311             )
  1312             if outcome == follower_contract.ReplayOutcome.replay:
! 1313                 return existing
  1314             # No updated_at on the wire for this op either — MLRun stamps its own
  1315             # local time for when it applied the change.
  1316             updated_at = datetime.datetime.now(datetime.UTC)
  1317             # The common set only — labels/annotations/owner/description. Everything else

  1388     def _decode_project_states_cursor(cursor: str) -> tuple[datetime.datetime, str]:
  1389         updated_at_str, _, name = cursor.partition("|")
  1390         try:
  1391             updated_at = datetime.datetime.fromisoformat(updated_at_str)
! 1392         except ValueError as exc:
! 1393             raise mlrun.errors.MLRunBadRequestError(
  1394                 f"Invalid list-states cursor: {cursor}"
  1395             ) from exc
  1396         return updated_at, name