workers: revision worker implementation

PREVIOUS versions (most recent first):
- 8e73e5a
- 403e760
- 262f7a5

WIP DO NOT MERGE
Commit message TBD

- add main worker flag and capacity/throttle flags
- add method to parse diff and list affected files
- add more test coverage for revision_worker.py
- add mots integration (bug 1740107)
- add new RevisionWorker that pre-processes revisions (bug 1788728)
- add new start/stop commands to manage workers
- add new flags to stop workers gracefully (*_WORKER_STOPPED)
- add repo.use_revision_worker feature flag (bug 1788732)
- add mots hashes check
- include new Lando revision info via API endpoint
- refactor dependency and stack fetching and parsing using networkx
- rename old command lando-cli landing-worker to lando-cli start-landing-worker
- run pre/post mots query
- store mots output in revision model

TODO:
- detect stack change on page load
- add tests for new warnings

.flake8

@@ -1,7 +1,7 @@
 [flake8]
 max-line-length = 88
 select = C,E,F,W,B,B9
-ignore = E203, E501, W503, B006
+ignore = E203, E501, W503, B006, E712, E711
 exclude =
     .hg,
     .git,

Dockerfile

@@ -54,8 +54,9 @@ RUN cd / && pip install --no-cache /app
 ENV PYTHONPATH /app
 RUN chown -R app:app /app
 
-# Create repos directory for transplanting in landing-worker
+# Create repos directory for landing-worker and revision worker.
 RUN mkdir /repos
+RUN chown -R app:app /repos
 
 # Run as a non-privileged user
 USER app

Dockerfile-dev

@@ -20,6 +20,7 @@ ENV PYTHONUNBUFFERED=1
 ENV FLASK_RUN_PORT=9000
 ENV FLASK_RUN_HOST=0.0.0.0
 ENV FLASK_DEBUG=1
+ENV HTTP_ALLOWED=1
 
 ENTRYPOINT ["lando-cli"]
 CMD ["run"]
@@ -48,6 +49,7 @@ RUN cd / && pip install --no-cache /app
 ENV PYTHONPATH /app
 RUN chown -R app:app /app
 
+# Create repos directory for landing worker and revision worker.
 RUN mkdir /repos
 RUN chown -R app:app /repos
 

docker-compose.yml

@@ -111,25 +111,24 @@ services:
       - smtp
   lando-api.landing-worker:
     image: lando-api
-    command: ["landing-worker"]
+    command: ["start-landing-worker"]
     environment:
-      - ENV=localdev
-      - DATABASE_URL=postgresql://postgres:[email protected]/lando_api_dev
-      - SENTRY_DSN=
-      # See http://docs.celeryproject.org/en/stable/getting-started/brokers/redis.html#configuration
-      # for the full URL format.
-      - CELERY_BROKER_URL=redis://redis.queue/0
-      - OIDC_IDENTIFIER=https://lando-api.test
-      - OIDC_DOMAIN=https://auth0.test
-      - LANDO_UI_URL=https://lando.test
-      - REPO_CLONES_PATH=/repos
-      - REPOS_TO_LAND=localdev
+      CELERY_BROKER_URL: "redis://redis.queue/0"
+      DATABASE_URL: "postgresql://postgres:[email protected]/lando_api_dev"
+      ENV: "localdev"
+      LANDO_UI_URL: "https://lando.test"
+      OIDC_DOMAIN: "https://auth0.test"
+      OIDC_IDENTIFIER: "https://lando-api.test"
+      REPOS_TO_LAND: "localdev"
+      REPO_CLONES_PATH: "/repos"
+      SENTRY_DSN: ""
     user: root
     volumes:
       - ./:/app
       - ./migrations/:/migrations/
       # Prevent writing python cache to the host.
       - caches_cache:/app/.cache/
+      - repos:/repos
     depends_on:
       - lando-api.db
       - redis.queue
@@ -157,3 +156,4 @@ volumes:
   caches_pycache:
   caches_cache:
   caches_pytest_cache:
+  repos:

landoapi/api/landing_jobs.py

@@ -63,6 +63,9 @@ def put(landing_job_id: str, data: dict):
 
     if landing_job.status in (LandingJobStatus.SUBMITTED, LandingJobStatus.DEFERRED):
         landing_job.transition_status(LandingJobAction.CANCEL)
+        for revision in landing_job.revisions:
+            # Unlock patches so they can be modified in the future.
+            revision.patch_locked = False
         db.session.commit()
         return {"id": landing_job.id}, 200
     else:

landoapi/api/transplants.py

@@ -365,7 +365,7 @@ def post(phab: PhabricatorClient, data: dict):
         }
 
         raw_diff = phab.call_conduit("differential.getrawdiff", diffID=diff["id"])
-        lando_revision.set_patch(raw_diff, patch_data)
+        lando_revision.set_patch(raw_diff, patch_data, final=True)
         db.session.commit()
         lando_revisions.append(lando_revision)
 
@@ -446,11 +446,9 @@ def get_list(phab: PhabricatorClient, stack_revision_id: str):
         limit=len(revision_phids),
     )
 
-    # Return both transplants and landing jobs, since for repos that were switched
-    # both or either of these could be populated.
-
     rev_ids = [phab.expect(r, "id") for r in phab.expect(revs, "data")]
 
+    # Find landing jobs based on related revisions or legacy revision_to_diff_id field.
     landing_jobs = LandingJob.revisions_query(rev_ids).all()
 
     return [job.serialize() for job in landing_jobs], 200

landoapi/cli.py

@@ -68,20 +68,64 @@ def worker(celery_arguments):
     celery.worker_main((sys.argv[0],) + celery_arguments)
 
 
-@cli.command(name="landing-worker")
-def landing_worker():
+@cli.command(name="start-landing-worker")
+def start_landing_worker():
     from landoapi.app import auth0_subsystem, lando_ui_subsystem
+    from landoapi.workers.landing_worker import LandingWorker
 
     exclusions = [auth0_subsystem, lando_ui_subsystem]
     for system in get_subsystems(exclude=exclusions):
         system.ensure_ready()
 
-    from landoapi.workers.landing_worker import LandingWorker
+    ConfigurationVariable.set(LandingWorker.STOP_KEY, VariableType.BOOL, "0")
 
     worker = LandingWorker()
     worker.start()
 
 
+@cli.command(name="stop-landing-worker")
+def stop_landing_worker():
+    from landoapi.workers.landing_worker import LandingWorker
+    from landoapi.storage import db_subsystem
+
+    db_subsystem.ensure_ready()
+    ConfigurationVariable.set(LandingWorker.STOP_KEY, VariableType.BOOL, "1")
+
+
+@cli.command(name="start-revision-worker")
+@click.argument("role")
+def start_revision_worker(role):
+    from landoapi.app import auth0_subsystem, lando_ui_subsystem
+    from landoapi.workers.revision_worker import RevisionWorker, Supervisor, Processor
+
+    roles = {
+        "processor": Processor,
+        "supervisor": Supervisor,
+    }
+
+    if role not in roles:
+        raise ValueError(f"Unknown worker role specified ({role}).")
+
+    exclusions = [auth0_subsystem, lando_ui_subsystem]
+    for system in get_subsystems(exclude=exclusions):
+        system.ensure_ready()
+
+    ConfigurationVariable.set(RevisionWorker.STOP_KEY, VariableType.BOOL, "0")
+
+    worker = roles[role]()
+    worker.start()
+
+
+@cli.command(name="stop-revision-worker")
+def stop_revision_worker():
+    """Stops all revision workers (supervisor and processors)."""
+    from landoapi.workers.revision_worker import RevisionWorker
+    from landoapi.storage import db_subsystem
+
+    db_subsystem.ensure_ready()
+    RevisionWorker.stop()
+
+
 @cli.command(name="run-pre-deploy-sequence")
 def run_pre_deploy_sequence():
     """Runs the sequence of commands required before a deployment."""

landoapi/commit_message.py

@@ -45,7 +45,7 @@
 IRC_NICK = r"[a-zA-Z0-9\-\_.]*[a-zA-Z0-9\-\_]+"
 
 # fmt: off
-REVIEWERS_RE = re.compile(  # noqa: E131
+REVIEWERS_RE = re.compile(
     r"([\s\(\.\[;,])"                   # before "r" delimiter
     + r"(" + SPECIFIER + r")"           # flag
     + r"("                              # capture all reviewers
@@ -209,3 +209,6 @@ def bug_list_to_commit_string(bug_ids: Iterable[str]) -> str:
         return "No bug"
 
     return f"Bug {', '.join(sorted(set(bug_ids)))}"
+
+
+# flake8: noqa: E131

landoapi/hg.py

@@ -1,8 +1,8 @@
 # This Source Code Form is subject to the terms of the Mozilla Public
 # License, v. 2.0. If a copy of the MPL was not distributed with this
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
-import copy
 import configparser
+import copy
 import logging
 import os
 import shlex
@@ -650,3 +650,13 @@ def read_checkout_file(self, path: str) -> str:
 
         with checkout_file_path.open() as f:
             return f.read()
+
+    def has_incoming(self, source: str) -> bool:
+        """Check if there are any incoming changes from the remote repo."""
+        try:
+            self.run_hg(["incoming", source, "--limit", "1"])
+        except hglib.error.CommandError as e:
+            if b"no changes found" not in e.out:
+                logger.error(e)
+            return False
+        return True

landoapi/models/__init__.py

@@ -3,11 +3,10 @@
 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
 from landoapi.models.landing_job import LandingJob
-from landoapi.models.revisions import Revision
 from landoapi.models.secapproval import SecApprovalRequest
 from landoapi.models.transplant import Transplant
 from landoapi.models.configuration import ConfigurationVariable
-from landoapi.models.revisions import DiffWarning
+from landoapi.models.revisions import DiffWarning, Revision
 
 __all__ = [
     "LandingJob",

landoapi/models/configuration.py

@@ -24,6 +24,9 @@ class ConfigurationKey(enum.Enum):
 
     LANDING_WORKER_PAUSED = "LANDING_WORKER_PAUSED"
     LANDING_WORKER_STOPPED = "LANDING_WORKER_STOPPED"
+    REVISION_WORKER_PAUSED = "REVISION_WORKER_PAUSED"
+    REVISION_WORKER_STOPPED = "REVISION_WORKER_STOPPED"
+    REVISION_WORKER_CAPACITY = "REVISION_WORKER_CAPACITY"
     API_IN_MAINTENANCE = "API_IN_MAINTENANCE"
     WORKER_THROTTLE_SECONDS = "WORKER_THROTTLE_SECONDS"
 

landoapi/models/landing_job.py

@@ -19,7 +19,7 @@
 from sqlalchemy.dialects.postgresql.json import JSONB
 
 from landoapi.models.base import Base
-from landoapi.models.revisions import Revision, revision_landing_job
+from landoapi.models.revisions import Revision, RevisionStatus, revision_landing_job
 from landoapi.storage import db
 
 logger = logging.getLogger(__name__)
@@ -38,7 +38,7 @@ class LandingJobStatus(enum.Enum):
     column of `LandingJob`.
     """
 
-    # Initial creation state.
+    # Ready to be picked up state.
     SUBMITTED = "SUBMITTED"
 
     # Actively being processed.
@@ -274,6 +274,14 @@ def set_landed_revision_diffs(self):
                 .values(diff_id=revision.diff_id)
             )
 
+    def has_non_ready_revisions(self) -> bool:
+        """Return whether any of the revisions are in a non-ready state or not."""
+        return bool(
+            set(r.status for r in self.revisions).intersection(
+                RevisionStatus.NON_READY_STATES
+            )
+        )
+
     def transition_status(
         self,
         action: LandingJobAction,
@@ -323,21 +331,42 @@ def transition_status(
 
         self.status = actions[action]["status"]
 
+        if action == LandingJobAction.CANCEL:
+            self.ready_revisions()
+
         if action in (LandingJobAction.FAIL, LandingJobAction.DEFER):
             self.error = kwargs["message"]
+            self.fail_revisions()
 
         if action == LandingJobAction.LAND:
             self.landed_commit_id = kwargs["commit_id"]
+            self.land_revisions()
 
         if commit:
             db.session.commit()
 
+    def fail_revisions(self):
+        """Mark all revisions in landing jobs as failed."""
+        for revision in self.revisions:
+            revision.fail()
+
+    def land_revisions(self):
+        """Mark all revisions in landing jobs as landed."""
+        for revision in self.revisions:
+            revision.land()
+
+    def ready_revisions(self):
+        """Mark all revisions in landing jobs as ready."""
+        for revision in self.revisions:
+            revision.ready()
+
     def serialize(self) -> dict[str, Any]:
         """Return a JSON compatible dictionary."""
         return {
             "id": self.id,
             "status": self.status.value,
             "landing_path": self.serialized_landing_path,
+            "duration_seconds": self.duration_seconds,
             "error_breakdown": self.error_breakdown,
             "details": (
                 self.error or self.landed_commit_id