Fix/publish concurrency

AUTHORS

@@ -60,3 +60,4 @@ List of contributors, in chronological order:
 * Nic Waller (https://github.com/sf-nwaller)
 * iofq (https://github.com/iofq)
 * Noa Resare (https://github.com/nresare)
+* Ramón N.Rodriguez (https://github.com/runitonmetal)

Makefile

@@ -53,7 +53,7 @@ ifeq ($(RUN_LONG_TESTS), yes)
 	go test -v -coverpkg="./..." -c -tags testruncli
 	if [ ! -e ~/aptly-fixture-db ]; then git clone https://github.com/aptly-dev/aptly-fixture-db.git ~/aptly-fixture-db/; fi
 	if [ ! -e ~/aptly-fixture-pool ]; then git clone https://github.com/aptly-dev/aptly-fixture-pool.git ~/aptly-fixture-pool/; fi
-	PATH=$(BINPATH)/:$(PATH) && . system/env/bin/activate && APTLY_VERSION=$(VERSION) $(PYTHON) system/run.py --long $(TESTS) --coverage-dir $(COVERAGE_DIR) $(CAPTURE)
+	PATH=$(BINPATH)/:$(PATH) && . system/env/bin/activate && APTLY_VERSION=$(VERSION) FORCE_COLOR=1 $(PYTHON) system/run.py --long $(TESTS) --coverage-dir $(COVERAGE_DIR) $(CAPTURE)
 endif
 
 docker-test: install
@@ -95,10 +95,16 @@ version:  ## Print aptly version
 docker-build-system-tests:  ## Build system-test docker image
 	docker build -f system/Dockerfile --no-cache . -t aptly-system-test
 
+docker-unit-tests:  ## Run unit tests in docker container
+	docker run -it --rm -v ${PWD}:/app aptly-system-test go test -v ./... -gocheck.v=true
+
 docker-system-tests:  ## Run system tests in docker container (add TEST=t04_mirror to run only specific tests)
-	docker run -t --rm -v ${PWD}:/app aptly-system-test /app/system/run-system-tests $(TEST)
+	docker run -it --rm -v ${PWD}:/app aptly-system-test /app/system/run-system-tests $(TEST)
 
 golangci-lint:  ## Run golangci-line in docker container
-	docker run -t --rm -v ~/.cache/golangci-lint/v1.56.2:/root/.cache -v ${PWD}:/app -w /app golangci/golangci-lint:v1.56.2 golangci-lint run
+	docker run -it --rm -v ~/.cache/golangci-lint/v1.56.2:/root/.cache -v ${PWD}:/app -w /app golangci/golangci-lint:v1.56.2 golangci-lint run
+
+flake8:
+	flake8 system
 
 .PHONY: help man modules version release goxc docker-build docker-system-tests

api/api.go

@@ -164,9 +164,17 @@
 		c.JSON(202, task)
 	} else {
 		log.Debug().Msg("Executing task synchronously")
-		out := context.Progress()
-		detail := task.Detail{}
-		retValue, err := proc(out, &detail)
+		task, conflictErr := runTaskInBackground(name, resources, proc)
+		if conflictErr != nil {
+			AbortWithJSONError(c, 409, conflictErr)
+			return
+		}
+
+		// wait for task to finish
+		context.TaskList().WaitForTaskByID(task.ID)
+
+		retValue, _ := context.TaskList().GetTaskReturnValueByID(task.ID)
+		err, _ := context.TaskList().GetTaskErrorByID(task.ID)
 		if err != nil {
 			AbortWithJSONError(c, retValue.Code, err)
 			return

context/context.go

@@ -567,6 +567,9 @@ func (context *AptlyContext) Shutdown() {
 			context.fileMemProfile = nil
 		}
 	}
+	if context.taskList != nil {
+		context.taskList.Stop()
+	}
 	if context.database != nil {
 		context.database.Close()
 		context.database = nil

system/Dockerfile

@@ -4,7 +4,7 @@ RUN apt-get update -y && apt-get install -y --no-install-recommends curl gnupg &
 
 RUN echo deb http://deb.debian.org/debian bookworm-backports main > /etc/apt/sources.list.d/backports.list
 RUN apt-get update && \
-    apt-get install -y --no-install-recommends apg bzip2 xz-utils ca-certificates golang/bookworm-backports golang-go/bookworm-backports golang-doc/bookworm-backports golang-src/bookworm-backports make git python3 python3-requests-unixsocket && \
+    apt-get install -y --no-install-recommends apg bzip2 xz-utils ca-certificates golang/bookworm-backports golang-go/bookworm-backports golang-doc/bookworm-backports golang-src/bookworm-backports make git python3 python3-requests-unixsocket python3-termcolor python3-swiftclient python3-boto && \
     apt-get clean && rm -rf /var/lib/apt/lists/*
 
 RUN useradd -m --shell /bin/sh --home-dir /var/lib/aptly aptly

system/lib.py

@@ -20,6 +20,7 @@
 import urllib.parse
 import urllib.request
 import zlib
+
 from pathlib import Path
 from uuid import uuid4
 
@@ -246,7 +247,7 @@ def prepare_fixture(self):
         if hasattr(self, "fixtureCmds"):
             for cmd in self.fixtureCmds:
                 output = self.run_cmd(cmd)
-                print("\n")
+                print("fixture Output:\n")
                 for line in output.decode("utf-8").split("\n"):
                     print(f"    {line}")
 

system/run-system-tests

@@ -4,7 +4,7 @@
 rm -rf /app/tmp
 rm -rf /tmp/aptly*
 
-usermod -u `stat -c %u /app` aptly
+usermod -u `stat -c %u /app` aptly >/dev/null
 chown -R `stat -c %u /app` /var/lib/aptly
 
 # use same /home/runner dir as in github workflow

system/run.py

@@ -33,14 +33,21 @@ def natural_key(string_):
     return [int(s) if s.isdigit() else s for s in re.split(r'(\d+)', string_)]
 
 
-def walk_modules(package):
-    yield importlib.import_module(package)
-    for name in sorted(glob.glob(package + "/*.py"), key=natural_key):
-        name = os.path.splitext(os.path.basename(name))[0]
-        if name == "__init__":
-            continue
+class TestStdout:
+    def __init__(self):
+        self.output = []
 
-        yield importlib.import_module(package + "." + name)
+    def write(self, text):
+        self.output.append(text)
+
+    def flush(self):
+        pass
+
+    def get_output(self):
+        return ''.join(self.output)
+
+    def truncate(self):
+        self.output = []
 
 
 def run(include_long_tests=False, capture_results=False, tests=None, filters=None, coverage_dir=None):
@@ -56,8 +63,22 @@ def run(include_long_tests=False, capture_results=False, tests=None, filters=Non
         coverage_dir = mkdtemp(suffix="aptly-coverage")
 
     for test in tests:
-        for testModule in walk_modules(test):
+        orig_stdout = sys.stdout
+        orig_stderr = sys.stderr
+
+        # importlib.import_module(test)
+        for name in sorted(glob.glob(test + "/*.py"), key=natural_key):
+            name = os.path.splitext(os.path.basename(name))[0]
+            if name == "__init__":
+                continue
+
+            testout = TestStdout()
+            sys.stdout = testout
+            sys.stderr = testout
+            testModule = importlib.import_module(test + "." + name)
+
             for name in sorted(dir(testModule), key=natural_key):
+                testout.truncate()
                 o = getattr(testModule, name)
 
                 if not (inspect.isclass(o) and issubclass(o, BaseTest) and o is not BaseTest and
@@ -81,17 +102,18 @@ def run(include_long_tests=False, capture_results=False, tests=None, filters=Non
                     if not matches:
                         continue
 
-                sys.stdout.write("%s:%s... " % (test, o.__name__))
-                sys.stdout.flush()
+                orig_stdout.write("%s: %s ... " % (test, colored(o.__name__, color="yellow", attrs=["bold"])))
+                orig_stdout.flush()
 
                 t = o()
+
                 if t.longTest and not include_long_tests or not t.fixture_available() or t.skipTest:
                     numSkipped += 1
                     msg = 'SKIP'
                     if t.skipTest and t.skipTest is not True:
                         # If we have a reason to skip, print it
                         msg += ': ' + t.skipTest
-                    sys.stdout.write(colored(msg + "\n", color="yellow"))
+                    orig_stdout.write(colored(msg + "\n", color="yellow"))
                     continue
 
                 numTests += 1
@@ -104,13 +126,18 @@ def run(include_long_tests=False, capture_results=False, tests=None, filters=Non
                     numFailed += 1
                     typ, val, tb = sys.exc_info()
                     fails.append((test, t, typ, val, tb, testModule))
-                    traceback.print_exception(typ, val, tb)
-                    sys.stdout.write(colored("FAIL\n", color="red"))
+                    orig_stdout.write(colored("\b\b\b\bFAIL\n", color="red", attrs=["bold"]))
+
+                    orig_stdout.write(testout.get_output())
+                    traceback.print_exception(typ, val, tb, file=orig_stdout)
                 else:
-                    sys.stdout.write(colored("OK\n", color="green"))
+                    orig_stdout.write(colored("\b\b\b\bOK \n", color="green", attrs=["bold"]))
 
                 t.shutdown()
 
+        sys.stdout = orig_stdout
+        sys.stderr = orig_stderr
+
     if lastBase is not None:
         lastBase.shutdown_class()
 

system/t12_api/publish.py

@@ -1,5 +1,6 @@
 import inspect
 import os
+import threading
 
 from api_lib import TASK_SUCCEEDED, APITest
 
@@ -282,6 +283,106 @@ def check(self):
         self.check_not_exists("public/" + prefix + "dists/")
 
 
+class PublishConcurrentUpdateAPITestRepo(APITest):
+    """
+    PUT /publish/:prefix/:distribution (local repos), DELETE /publish/:prefix/:distribution
+    """
+    fixtureGpg = True
+
+    def check(self):
+        repo_name = self.random_name()
+        self.check_equal(self.post(
+            "/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
+
+        d = self.random_name()
+        self.check_equal(
+            self.upload("/api/files/" + d,
+                        "pyspi_0.6.1-1.3.dsc",
+                        "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
+                        "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
+        self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED)
+
+        prefix = self.random_name()
+        resp = self.post_task(
+            "/api/publish/" + prefix,
+            json={
+                "Architectures": ["i386", "source"],
+                "SourceKind": "local",
+                "Sources": [{"Name": repo_name}],
+                "Signing": DefaultSigningOptions,
+            }
+        )
+
+        self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
+
+        self.check_not_exists(
+            "public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
+        self.check_exists("public/" + prefix +
+                          "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
+
+        d = self.random_name()
+        self.check_equal(self.upload("/api/files/" + d,
+                         "libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
+        self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED)
+
+        self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/",
+                         json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED)
+
+        def _do_update(result, index):
+            resp = self.put_task(
+                "/api/publish/" + prefix + "/wheezy",
+                json={
+                    "AcquireByHash": True,
+                    "Signing": DefaultSigningOptions,
+                }
+            )
+            try:
+                self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
+            except BaseException as e:
+                result[index] = e
+
+        n_workers = 10
+        worker_results = [None] * n_workers
+        tasks = [threading.Thread(target=_do_update, args=(worker_results, i,)) for i in range(n_workers)]
+        [task.start() for task in tasks]
+        [task.join() for task in tasks]
+        for result in worker_results:
+            if isinstance(result, BaseException):
+                raise result
+
+        repo_expected = {
+            'AcquireByHash': True,
+            'Architectures': ['i386', 'source'],
+            'Codename': '',
+            'Distribution': 'wheezy',
+            'Label': '',
+            'Origin': '',
+            'NotAutomatic': '',
+            'ButAutomaticUpgrades': '',
+            'Path': prefix + '/' + 'wheezy',
+            'Prefix': prefix,
+            'SkipContents': False,
+            'SourceKind': 'local',
+            'Sources': [{'Component': 'main', 'Name': repo_name}],
+            'Storage': '',
+            'Suite': ''}
+
+        all_repos = self.get("/api/publish")
+        self.check_equal(all_repos.status_code, 200)
+        self.check_in(repo_expected, all_repos.json())
+
+        self.check_exists("public/" + prefix +
+                          "/dists/wheezy/main/binary-i386/by-hash")
+
+        self.check_exists(
+            "public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
+        self.check_not_exists(
+            "public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
+
+        self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy").json()['State'], TASK_SUCCEEDED)
+        self.check_not_exists("public/" + prefix + "dists/")
+
+
 class PublishUpdateSkipCleanupAPITestRepo(APITest):
     """
     PUT /publish/:prefix/:distribution (local repos), DELETE /publish/:prefix/:distribution

system/t12_api/tasks.py

@@ -20,9 +20,9 @@ def _create_mirror(self, dist):
         resp = self.put("/api/mirrors/" + mirror_name, json=mirror_desc, params={'_async': True})
         self.check_equal(resp.status_code, 202)
 
-        # check that two mirror updates cannot run at the same time
+        # check that two mirror updates are queuedd
         resp2 = self.put("/api/mirrors/" + mirror_name, json=mirror_desc, params={'_async': True})
-        self.check_equal(resp2.status_code, 409)
+        self.check_equal(resp2.status_code, 202)
 
         return resp.json()['ID'], mirror_name