Merge ubuntu-debuginfod:concurrency-rework into ubuntu-debuginfod:master

[Sergio Durigan Junior]

This is a rework of the entire application after discussing with Robie and finding out that it did a poor job at treating race conditions.

The main idea behind this MP is that we will now keep the state of each task in a SQL database. I chose PostgreSQL because (a) I like the database, and (b) it does a very decent job a handling concurrent requests. Unfortunately, sqlite did not meet the standards I was looking for.

The poller (poll_launchpad.py) will poll Launchpad, get the list of ddebs and source packages that have been published since it last ran, and dispatch tasks for each one. While doing so, it will also update the database and insert entries for each task. The poller is a single process.

The getter is where things get more interesting. Celery will spawn multiple concurrent threads that will treat tasks as they arrive, and in an ideal world these tasks are all unique. However, there are certain scenarios (very unlikely to happen, but not impossible) where we might have duplicate tasks in the queue. Even in this scenario, as long as these tasks are processed in different moments everything should be fine, because the getter is idempotent. However^2, in the extremely unlikely scenario where two workers decide to process two identical tasks, we might have some race conditions when, e.g., writing the ddeb to disk. For this reason, the getter now tries to move the files atomically.

The Celery application (debuginfod.py) will monitor the tasks and update the SQL database accordingly. If the task succeeds, we don't need to care about it anymore and can simply remove its entry. If the task gets rescheduled (because of a timeout issue, for example), then we keep its entry intact, which should prevent duplicate tasks from being dispatched by the poller. There's also some error logging as well, because sometimes a task may throw an unexpected exception and we need to be able to log this problem in order to come back to it later. The logging is also done using a SQL table.

I ran extensive tests locally to make sure things are working correctly, and fixed every problem I found. I'm confident that this can be put in production, but I would like your input first. The MP is big but I split the commits logically, which should make things easier to review. Let me know if you need any clarification.

[Bryce Harrington (bryce)]

Quick review limited to just commit 8595463... yes this looks good. Since that commit is followup to the prior MP feel free to land it independently if you'd like.

Rest of this MP will take longer to digest.

[Bryce Harrington (bryce)]

Another quick look, just at the database code, with a refactoring suggestion.

[Sergio Durigan Junior (sergiodj)]

BTW, with my latest force-push I've also implemented a new class (DebugDB) which abstracts the access to the database. I think it's cleaner this way.

[Lena Voytek (lvoytek)]

So far this looks good to me. I've added some suggestions for cleanup below

[Sergio Durigan Junior (sergiodj)]

Thanks, Lena.

Addressed all comments and pushed a new commit on top of everything.

[Athos Ribeiro (athos-ribeiro)]

Looking good!

I added a few inline comments

review: Needs Information

[Sergio Durigan Junior (sergiodj)]

Thanks, Athos! Addressed most of your comments and pushed a new commit.

[Athos Ribeiro (athos-ribeiro)]

Thanks!

review: Approve

[Sergio Durigan Junior (sergiodj)]

Thanks for the reviews, folks.

diff --git a/README.md b/README.md
index 8fe7f00..3c49598 100644
--- a/README.md
+++ b/README.md
@@ -8,18 +8,21 @@ This project is used in the deployment of Ubuntu's
 * Python
 
 ```
-python3-pyelftools >= 0.29-1
-python3-debian
 python3-git
 python3-celery
 python3-launchpadlib
 python3-requests
 python3-sdnotify
+python3-psycopg2
 ```
 
 * Applications
 
 ```
 celery
-rabbitmq-server
+rabbitmq-server*
+postgresql-server*
+
+*: These could be charms, but (as of this writing) they don't yet
+   support Jammy.
 ```
diff --git a/ddebgetter.py b/ddebgetter.py
index 62eebc6..661d477 100644
--- a/ddebgetter.py
+++ b/ddebgetter.py
@@ -18,16 +18,11 @@
 # Authors: Sergio Durigan Junior <[e-mail]>
 
 import os
-from urllib import parse
 import lzma
 import tarfile
 import subprocess
 
-from debian import debfile
 import tempfile
-from elftools.common.utils import bytelist2string
-from elftools.common.exceptions import ELFError, ELFParseError, DWARFError
-from elftools.elf.elffile import ELFFile
 
 from git import Git, Repo
 from git.exc import GitCommandError
@@ -49,45 +44,42 @@ class DdebGetter(DebugGetter):
 
         :param request: The dictionary containing the information
             necessary to fetch this ddeb.
-        :type request: dict(str : str)
-        """
-        if not request or not request.get("ddebs"):
-            return
-
-        self._logger.debug(f"Processing request to download ddebs: {request}")
-        self._download_ddebs(
-            request["source_package"],
-            request["component"],
-            request["ddebs"],
+        :type request: dict(str : str)"""
+        if request is None:
+            raise TypeError("Invalid request (None)")
+        if request.get("ddeb_url") is None:
+            raise TypeError("No 'ddeb_url' in request")
+        if request.get("ddeb_url") is None:
+            raise TypeError("No 'architecture' in request")
+        if request["architecture"] == "source":
+            raise ValueError("Wrong request: source fetch")
+
+        self._logger.debug(f"Processing request to download ddeb: {request}")
+        self._download_ddeb(
+            request["source_package"], request["component"], request["ddeb_url"]
         )
 
-    def _download_ddebs(self, source_package, component, urls):
-        """Download the ddebs associated with a package/version.
+    def _download_ddeb(self, source_package, component, ddeb_url):
+        """Download a ddeb associated with a package.
 
         :param str source_package: Source package name.
 
-        :param str version: Source package version.
-
         :param str component: Source package component.
 
-        :param list urls: List of ddeb URLs."""
+        :param str ddeb_url: The ddeb URL."""
         savepath = self._make_savepath(source_package, component)
-        os.makedirs(savepath, mode=0o755, exist_ok=True)
-        for url in urls:
-            self._logger.debug(f"Downloading '{url}' into '{savepath}'")
-            self._download_from_lp(url, savepath)
+        self._logger.debug(f"Downloading '{ddeb_url}' into '{savepath}'")
+        self._download_from_lp(ddeb_url, savepath)
 
 
 # The function below was taken from git-ubuntu.  We need to make
 # sure we perform the same version -> tag transformation as it
 # does.
-def _git_dep14_tag(version):
+def git_dep14_tag(version):
     """Munge a version string according to http://dep.debian.net/deps/dep14/
 
     :param str version: The version to be adjusted."""
-    version = version.replace("~", "_")
-    version = version.replace(":", "%")
-    version = version.replace("..", ".#.")
+    version = version.replace("~", "_").replace(":", "%").replace("..", ".#.")
     if version.endswith("."):
         version = version + "#"
     if version.endswith(".lock"):
@@ -95,6 +87,20 @@ def _git_dep14_tag(version):
         version = pre + ".#lock"
     return version
 
+def adjust_tar_filepath(tarinfo):
+    """Adjust the filepath for a TarInfo file.
+
+    This function is needed because TarFile.add strips the leading
+    slash from the filenames, so we have to workaround it by
+    re-adding the slash ourselves.
+
+    This function is intended to be used as a callback provided to
+    TarFile.add.
+
+    :param TarInfo tarinfo: The tarinfo."""
+    tarinfo.name = os.path.join("/", tarinfo.name)
+    return tarinfo
+
 
 class DdebSourceCodeGetter(DebugGetter):
     """Get (fetch) the source code associated with a ddeb."""
@@ -107,113 +113,24 @@ class DdebSourceCodeGetter(DebugGetter):
 
         :param request: The dictionary containing the information
             necessary to fetch this source code.
-        :type request: dict(str : str)
-        """
-        if not request or not request.get("ddebs"):
-            return
-
-        self._logger.debug(
-            f"Processing request to download source code (for ddebs): {request}"
-        )
+        :type request: dict(str : str)"""
+        if request is None:
+            raise TypeError("Invalid request (None)")
+        if request.get("source_urls") is None:
+            raise TypeError("No 'source_urls' in request")
+        if request.get("ddeb_url") is None:
+            raise TypeError("No 'architecture' in request")
+        if request["architecture"] != "source":
+            raise ValueError("Wrong request: ddeb fetch")
+
+        self._logger.debug(f"Processing request to download source code: {request}")
         self._process_source(
             request["source_package"],
             request["version"],
             request["component"],
-            request["ddebs"],
-            request["sources"],
+            request["source_urls"],
         )
 
-    def _get_comp_dirs(self, debug_file, filename):
-        """Get the list of available DW_AT_comp_dir declarations from the
-        DWARF file.
-
-        :param TextIOWrapper debug_file: The .debug file.
-        """
-        elf_file = ELFFile(debug_file)
-
-        if not elf_file.has_dwarf_info():
-            self._logger.debug(f"'{filename}' doesn't have DWARF")
-            return []
-
-        result = []
-        dwarf_file = elf_file.get_dwarf_info()
-        for cu in dwarf_file.iter_CUs():
-            attr = cu.get_top_DIE().attributes
-            if "DW_AT_comp_dir" in attr.keys():
-                value = attr["DW_AT_comp_dir"].value
-                if isinstance(value, bytes):
-                    result.append(bytelist2string(value).decode("UTF-8"))
-        self._logger.debug(
-            f"Found the following DW_AT_comp_dir for '{filename}':\n{result}"
-        )
-        return result
-
-    def _should_fetch_source_for_debugfile(self, source_package, version, filepath):
-        """Whether we should fetch the source code for the specified source
-        package/deb.
-
-        We return True if the debug file associated with the source
-        package has any DW_AT_comp_dir entity whose path points to
-        "/usr/src/source_package-version/".
-
-        :param str source_package: Source package name.
-
-        :param str version: Source package version.
-
-        :param str filepath: Full path to the .ddeb file."""
-        deb_file = debfile.DebFile(filepath)
-        # The path we expect to find in any of the DW_AT_comp_dirs.
-        expected_path = f"/usr/src/{source_package}-{version}/"
-        for deb_internal_file in list(deb_file.data):
-            if not deb_internal_file.endswith(".debug"):
-                continue
-            with tempfile.TemporaryFile() as debug_file:
-                debug_file.write(deb_file.data.get_content(deb_internal_file))
-                try:
-                    comp_dirs = self._get_comp_dirs(debug_file, deb_internal_file)
-                except (ELFError, ELFParseError, DWARFError) as e:
-                    self._logger.warning(
-                        f"Exception while looking for DW_AT_comp_dirs: {e}"
-                    )
-                    # There's no point in retrying anything here, so
-                    # we just generate a warning and keep going.
-                    continue
-
-                for comp_dir in comp_dirs:
-                    if comp_dir.startswith(expected_path):
-                        self._logger.debug(f"Found a good DW_AT_comp_dir: {comp_dir}")
-                        return True
-        return False
-
-    def _should_fetch_source(self, source_package, version, component, filenames):
-        """Return True if we should fetch the source code for a specific
-        package, False otherwise.
-
-        Whether or not we should fetch the source code is determined
-        by the presence of a specific path prefix (/usr/src/...) in
-        the DW_AT_comp_dir declarations of a DWARF file.
-
-        :param str source_package: Source package name.
-
-        :param str version: Source package version.
-
-        :param str component: Source package component (main,
-            universe, etc.)
-
-        :param list urls: is the list of URLs used to fetch the respective debug
-            information for the package.
-        """
-        savepath = self._make_savepath(source_package, component)
-        for fname in filenames:
-            filepath = os.path.join(savepath, fname)
-            if not os.path.isfile(filepath):
-                continue
-            if self._should_fetch_source_for_debugfile(
-                source_package, version, filepath
-            ):
-                return True
-        return False
-
     def _download_source_code_from_git(self, source_package, version, filepath):
         """Download the source code using Launchpad's git repository.
 
@@ -223,11 +140,16 @@ class DdebSourceCodeGetter(DebugGetter):
 
         :param str filepath: The full pathname where the resulting
             source code tarball should be saved.
-        """
+
+        :rtype: bool
+
+        This method returns True when the operation succeeds, or False
+        *iff* the "git clone" command fails to run.  Otherwise, this
+        function will throw an exception."""
         with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as source_dir:
             g = Git()
             git_dir = os.path.join(source_dir, f"{source_package}")
-            tag = "applied/" + _git_dep14_tag(f"{version}")
+            tag = "applied/" + git_dep14_tag(f"{version}")
             self._logger.debug(
                 f"Cloning '{source_package}' git repo into '{git_dir}' (tag: '{tag}')"
             )
@@ -247,29 +169,16 @@ class DdebSourceCodeGetter(DebugGetter):
                 return False
             repo = Repo(git_dir)
             prefix_path = os.path.join("/usr/src/", f"{source_package}-{version}/")
-            with lzma.open(filepath, "w") as xzfile:
-                self._logger.debug(
-                    f"Archiving git repo for '{source_package}-{version}' as '{filepath}'"
-                )
-                repo.archive(xzfile, prefix=prefix_path, format="tar")
+            self._logger.debug(
+                f"Archiving git repo for '{source_package}-{version}' as '{filepath}'"
+            )
+            with tempfile.NamedTemporaryFile() as tmpfile:
+                with lzma.open(tmpfile.name, "w") as xzfile:
+                    repo.archive(xzfile, prefix=prefix_path, format="tar")
+                self._try_to_move_atomically(tmpfile.name, filepath)
 
         return True
 
-    def _adjust_tar_filepath(self, tarinfo):
-        """Adjust the filepath for a TarInfo file.
-
-        This function is needed because TarFile.add strips the leading
-        slash from the filenames, so we have to workaround it by
-        re-adding the slash ourselves.
-
-        This function is intended to be used as a callback provided to
-        TarFile.add.
-
-        :param TarInfo tarinfo: The tarinfo.
-        """
-        tarinfo.name = f"/{tarinfo.name}"
-        return tarinfo
-
     def _download_source_code_from_dsc(
         self, source_package, version, filepath, source_urls
     ):
@@ -286,17 +195,30 @@ class DdebSourceCodeGetter(DebugGetter):
         :param list source_urls: List of URLs used to fetch the source
             package.  This is usually the list returned by the
             sourceFileUrls() Launchpad API call.
-        """
+
+        :rtype: bool
+        This method returns True when the operation succeeds, or False
+        *iff* the "dpkg-source -x" command fails to run.  Otherwise,
+        this function will throw an exception."""
         with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as source_dir:
             for url in source_urls:
                 self._download_from_lp(url, source_dir)
 
-            dscfile = os.path.join(source_dir, f"{source_package}_{version}.dsc")
-            if not os.path.isfile(dscfile):
-                self._logger.warning(f"'{dscfile}' doesn't exist, but it should.")
+            dscfile = None
+            for f in os.listdir(source_dir):
+                newf = os.path.join(source_dir, f)
+                if os.path.isfile(newf) and f.endswith(".dsc"):
+                    dscfile = newf
+                    break
+
+            if dscfile is None:
+                self._logger.warning(
+                    "Could not find .dsc file, even though it should exist."
+                )
                 return False
 
             outdir = os.path.join(source_dir, "outdir")
+            self._logger.debug(f"Will call 'dpkg-source -x {dscfile} {outdir}'")
             try:
                 subprocess.run(
                     ["/usr/bin/dpkg-source", "-x", dscfile, outdir],
@@ -315,8 +237,12 @@ class DdebSourceCodeGetter(DebugGetter):
 
             prefix_path = os.path.join("/usr/src/", f"{source_package}-{version}/")
 
-            with tarfile.open(filepath, "w:xz") as tfile:
-                tfile.add(outdir, arcname=prefix_path, filter=self._adjust_tar_filepath)
+            with tempfile.NamedTemporaryFile() as tmpfile:
+                with tarfile.open(tmpfile.name, "w:xz") as tfile:
+                    tfile.add(
+                        outdir, arcname=prefix_path, filter=adjust_tar_filepath
+                    )
+                self._try_to_move_atomically(tmpfile.name, filepath)
 
             return True
 
@@ -355,29 +281,14 @@ class DdebSourceCodeGetter(DebugGetter):
         # like to retry the task.
         raise DebugGetterRetry()
 
-    def _process_source(
-        self, source_package, version, component, ddeb_urls, source_urls
-    ):
+    def _process_source(self, source_package, version, component, source_urls):
         """Process the request to fetch the source code.
 
-        We only download the source code if we know it will be
-        properly indexed by debuginfod.
-
         :param str source_package: Source package name.
 
         :param str version: Source package version.
 
         :param str component: Source package component.
 
-        :param list urls: List of ddeb URLs."""
-        # Convert URLs into filenames.
-        filenames = [
-            os.path.basename(parse.urlparse(fname).path) for fname in ddeb_urls
-        ]
-        if not self._should_fetch_source(source_package, version, component, filenames):
-            self._logger.info(
-                f"Should not fetch source code for '{source_package}-{version}'"
-            )
-            return
-
+        :param list source_urls: List of source URLs."""
         self._download_source_code(source_package, version, component, source_urls)
diff --git a/ddebpoller.py b/ddebpoller.py
index 24ba613..6f92c2f 100644
--- a/ddebpoller.py
+++ b/ddebpoller.py
@@ -17,14 +17,18 @@
 
 # Authors: Sergio Durigan Junior <[e-mail]>
 
+import os
+from urllib import parse
 from debugpoller import DebugPoller
 
 
 class DdebPoller(DebugPoller):
-    def __init__(
-        self, initial_interval=None, force_initial_interval=False, dry_run=False
-    ):
-        """Initialize the object using 'ddeb' as its name.
+    """Perform the Launchpad polling and obtain a list of ddebs and source
+    packages."""
+
+    def __init__(self, initial_interval=1, force_initial_interval=False, dry_run=False):
+
+    """Initialize the object using 'ddeb' as its name.
 
         Look at DebugPoller's docstring for an explanation about the arguments."""
         super().__init__(
@@ -34,43 +38,90 @@ class DdebPoller(DebugPoller):
             dry_run=dry_run,
         )
 
-    def get_ddebs(self):
-        """Get the list of ddebs that have been published since the last
-        timestamp from Launchpad."""
+    def get_ddebs_and_sources(self):
+        """Get the list of ddebs and source files that have been published
+        since the last timestamp from Launchpad.
+
+        :rtype: dict, str
+
+        Return a dictionary containing all ddebs and source packages
+        found, and also the new timestamp that should then be recorded
+        by calling record_timestamp."""
         timestamp = self._get_timestamp()
-        assert timestamp is not None, f"_get_timestamp returned None"
-        assert timestamp != "", f"_get_timestamp returned a blank timestamp"
+        if timestamp is None:
+            raise RuntimeError("_get_timestamp returned None")
+        if timestamp == "":
+            raise RuntimeError("_get_timestamp returned a blank timestamp")
         new_timestamp = self._generate_timestamp()
-        assert new_timestamp is not None, f"_generate_timestamp returned None"
-        assert new_timestamp != "", f"_generate_timestamp returned a blank timestamp"
-
-        result = []
+        if new_timestamp is None:
+            raise RuntimeError("_generate_timestamp returned None")
+        if new_timestamp == "":
+            raise RuntimeError("_generate_timestamp returned a blank timestamp")
 
         self._logger.info(
             f"Polling ddebs created since '{timestamp}' (it's now '{new_timestamp}')"
         )
 
-        for pkg in self._main_archive.getPublishedSources(
-            order_by_date=True, created_since_date=timestamp, status="Published"
+        result = []
+        for pkg in self._main_archive.getPublishedBinaries(
+            order_by_date=True, created_since_date=timestamp
         ):
-            ddeb_urls = [url for url in pkg.binaryFileUrls() if url.endswith(".ddeb")]
-            ddebs_len = len(ddeb_urls)
-            if ddebs_len > 0:
-                distro_series = self._lp.load(pkg.distro_series_link).name
-                pkgname = pkg.source_package_name
-                pkgver = pkg.source_package_version
+            if pkg.status not in ("Pending", "Published"):
+                continue
+            if not pkg.is_debug:
+                continue
+
+            ddeb_urls = pkg.binaryFileUrls()
+            if len(ddeb_urls) == 0:
+                # Safety check.
+                continue
+
+            srcname = pkg.source_package_name
+            srcver = pkg.source_package_version
+            component = pkg.component_name
+
+            # We create one message (which will eventually become a
+            # Celery task) per ddeb.  This makes it easier later to do
+            # deduplication, and also has the benefit of making the
+            # downloading process more granular for multiple ddebs.
+            for url in ddeb_urls:
+                # Obtain the ddeb filename and its architecture.
+                ddeb = os.path.basename(parse.urlparse(url).path)
+                _, _, arch_and_extension = ddeb.split("_")
+                arch, _ = arch_and_extension.split(".")
+
                 msg = {
-                    "source_package": pkgname,
-                    "version": pkgver,
-                    "component": pkg.component_name,
-                    "distro_series": distro_series,
-                    "ddebs": ddeb_urls,
-                    "sources": pkg.sourceFileUrls(),
+                    "source_package": srcname,
+                    "version": srcver,
+                    "component": component,
+                    "ddeb_url": url,
+                    "ddeb_filename": ddeb,
+                    "architecture": arch,
                 }
-                self._logger.debug(
-                    f"For source package '{pkgname}-{pkgver}', found {ddebs_len}:\n{ddeb_urls}"
-                )
                 result.append(msg)
 
-        self._record_timestamp(new_timestamp)
-        return result
+        for pkg in self._main_archive.getPublishedSources(
+            order_by_date=True, created_since_date=timestamp
+        ):
+            if pkg.status not in ("Pending", "Published"):
+                continue
+
+            src_urls = pkg.sourceFileUrls()
+            if len(src_urls) == 0:
+                # Safety check.
+                continue
+
+            srcname = pkg.source_package_name
+            srcver = pkg.source_package_version
+            component = pkg.component_name
+
+            msg = {
+                "source_package": srcname,
+                "version": srcver,
+                "component": component,
+                "source_urls": src_urls,
+                "architecture": "source",
+            }
+            result.append(msg)
+
+        return result, new_timestamp
diff --git a/debuggetter.py b/debuggetter.py
index 9bb6102..bafa914 100644
--- a/debuggetter.py
+++ b/debuggetter.py
@@ -21,6 +21,8 @@ import os
 import requests
 from urllib import parse
 import logging
+import tempfile
+import shutil
 
 
 class DebugGetterTimeout(Exception):
@@ -34,6 +36,7 @@ class DebugGetterRetry(Exception):
 
 DEFAULT_MIRROR_DIR = "/srv/debug-mirror"
 
+
 class DebugGetter:
     """Base class for a Debug Getter."""
 
@@ -45,8 +48,7 @@ class DebugGetter:
 
         :param str subdir: The subdirectory (insider mirror_dir) where
             the module will save its files.  For example, a ddeb
-            getter module should specify "ddebs" here.
-        """
+            getter module should specify "ddebs" here."""
         self._mirror_dir = mirror_dir
         self._subdir = subdir
         self._logger = logging.getLogger(__name__)
@@ -55,17 +57,52 @@ class DebugGetter:
         """Return the full save path for a package.
 
         :param str source_package: The package name.
-        :param str component: The component name (main, universe, etc.).
-        """
+        :param str component: The component name (main, universe, etc.)."""
+        if source_package.startswith("lib"):
+            pkgname_initials = source_package[:4]
+        else:
+            pkgname_initials = source_package[0]
+
         return os.path.join(
             self._mirror_dir,
             self._subdir,
             component,
-            source_package[:1],
+            pkgname_initials,
             source_package,
         )
 
+    def _try_to_move_atomically(self, src, dst):
+        """Try to move SRC to DST atomically.
+
+        This function assumes that SRC and DST have different filenames.
+
+        :param str src: The source file (full path).
+
+        :param str dst: The destination file (full path)."""
+        self._logger.debug(f"Trying to move '{src}' to '{dst}' atomically")
+        savepath = os.path.dirname(dst)
+        os.makedirs(savepath, mode=0o755, exist_ok=True)
+        newtmpfile = os.path.join(savepath, os.path.basename(src))
+        shutil.copyfile(src, newtmpfile)
+        try:
+            shutil.move(newtmpfile, dst)
+        except shutil.SameFileError:
+            os.remove(newtmpfile)
+            self._logger.warning(
+                f"Could not create '{dst}' atomically: same file already exists"
+            )
+
     def _download_from_lp(self, url, savepath):
+        """Download a file from Launchpad.
+
+        This method tries to download the file in chunks into a
+        temporary file and then does an atomic move to the final
+        destination.
+
+        :param str url: The URL that should be downloaded.
+
+        :param str savepath: The full path (minus the filename) where
+        the file should be saved."""
         filepath = os.path.join(savepath, os.path.basename(parse.urlparse(url).path))
         if os.path.exists(filepath):
             self._logger.debug(f"'{filepath}' exists, doing nothing")
@@ -75,10 +112,12 @@ class DebugGetter:
             with requests.Session() as s:
                 with s.get(url, allow_redirects=True, timeout=10, stream=True) as r:
                     r.raise_for_status()
-                    with open(filepath, "wb") as f:
+                    with tempfile.NamedTemporaryFile(mode="wb") as tmpfile:
                         # 10 MB for chunk_size should be enough...
                         for chunk in r.iter_content(chunk_size=10 * 1024 * 1024):
-                            f.write(chunk)
+                            tmpfile.write(chunk)
+                        tmpfile.flush()
+                        self._try_to_move_atomically(tmpfile.name, filepath)
         except (
             requests.ConnectionError,
             requests.HTTPError,
diff --git a/debuginfod.py b/debuginfod.py
index c0d0357..5c5d77f 100644
--- a/debuginfod.py
+++ b/debuginfod.py
@@ -18,11 +18,16 @@
 # Authors: Sergio Durigan Junior <[e-mail]>
 
 from celery import Celery
-from celery.signals import worker_ready, worker_shutting_down
+from celery.signals import (
+    worker_ready,
+    worker_shutting_down,
+    task_postrun,
+    celeryd_init,
+)
 
 from debuggetter import DebugGetterTimeout, DebugGetterRetry
-
 from ddebgetter import DdebGetter, DdebSourceCodeGetter
+from utils import DebugDB
 
 import sdnotify
 
@@ -45,6 +50,23 @@ app.conf.update(
 sdnotifier = sdnotify.SystemdNotifier()
 
 
+def _record_error_into_db(exc, msg):
+    """Record an error into the database.
+
+    :param Exception exc: The Exception that was triggered.
+
+    :param dict msg: The message that triggere the error."""
+    with DebugDB() as db:
+        with db.cursor() as cur:
+            cur.execute(
+                "INSERT INTO errors (task, exception) VALUES (%s, %s)",
+                (
+                    str(msg),
+                    str(exc),
+                ),
+            )
+
+
 @app.task(
     name="grab_ddebs",
     autoretry_for=(DebugGetterTimeout, DebugGetterRetry),
@@ -58,10 +80,20 @@ def grab_ddebs(msg):
     """Dispatch the DdebGetter task.
 
     :param dict(str -> str) msg: The dictionary containing the message
-        that will be processed by the getter.
-    """
-    g = DdebGetter()
-    g.process_request(msg)
+        that will be processed by the getter."""
+    try:
+        g = DdebGetter()
+        g.process_request(msg)
+    except Exception as e:
+        if not isinstance(e, DebugGetterRetry) and not isinstance(
+            e, DebugGetterTimeout
+        ):
+            # This is some other kind of error that we need to deal
+            # with.  Mark it as such.
+            _record_error_into_db(e, msg)
+        # We still need to raise the exception here.  Celery will
+        # reschedule the task if applicable.
+        raise e
 
 
 @app.task(
@@ -77,10 +109,20 @@ def grab_ddebs_sources(msg):
     """Dispatch the DdebSourceCodeGetter task.
 
     :param dict(str -> str) msg: The dictionary containing the message
-        that will be processed by the getter.
-    """
-    g = DdebSourceCodeGetter()
-    g.process_request(msg)
+        that will be processed by the getter."""
+    try:
+        g = DdebSourceCodeGetter()
+        g.process_request(msg)
+    except Exception as e:
+        if not isinstance(e, DebugGetterRetry) and not isinstance(
+            e, DebugGetterTimeout
+        ):
+            # This is some other kind of error that we need to deal
+            # with.  Mark it as such.
+            _record_error_into_db(e, msg)
+        # We still need to raise the exception here.  Celery will
+        # reschedule the task if applicable.
+        raise e
 
 
 @worker_ready.connect
@@ -95,5 +137,38 @@ def notify_worker_shutting_down(**kwargs):
     sdnotifier.notify("STOPPING=1")
 
 
+@task_postrun.connect
+def task_postrun(sender, args, state, **kwargs):
+    """Process a task that's just finished.
+
+    We keep the state of each task in a SQL database, and this
+    function is responsible for cleaning up the state for tasks that have
+    succeeded."""
+    if state not in ("SUCCESS",):
+        return
+
+    with DebugDB() as db:
+        with db.cursor() as cur:
+            msg = args[0]
+            jobid = msg["jobid"]
+            cur.execute("DELETE FROM tasks WHERE id = %s", (jobid,))
+
+
+@celeryd_init.connect
+def setup_worker(sender=None, **kwargs):
+    """Setup the worker.
+
+    We have to create the tables that will be populated when
+    processing tasks."""
+    with DebugDB() as db:
+        with db.cursor() as cur:
+            cur.execute(
+                "CREATE TABLE IF NOT EXISTS tasks (id serial PRIMARY KEY, source text, version text, arch text)"
+            )
+            cur.execute(
+                "CREATE TABLE IF NOT EXISTS errors (id serial PRIMARY KEY, task text, exception text)"
+            )
+
+
 if __name__ == "__main__":
     app.start()
diff --git a/debugpoller.py b/debugpoller.py
index 058909d..2d74723 100644
--- a/debugpoller.py
+++ b/debugpoller.py
@@ -58,9 +58,10 @@ class DebugPoller:
 
         :param bool dry_run: Tell the poller that it shouldn't record
             the timestamp in the file when the operation finishes.
-            Default is False.
-        """
-        self._lp = Launchpad.login_anonymously("ubuntu-debuginfod poller", "production")
+            Default is False."""
+        self._lp = Launchpad.login_anonymously(
+            "ubuntu-debuginfod poller", "production", version="devel"
+        )
         self._main_archive = self._lp.distributions["ubuntu"].main_archive
         self.TIMESTAMP_FILE = self.TIMESTAMP_FILE + f"-{module_name}"
 
@@ -84,8 +85,7 @@ class DebugPoller:
         :param interval: Specify how long ago (in hours) the timestamp must
             refer to.  If not specified, the timestamp is generated for
             the current time.
-        :type interval: int or None
-        """
+        :type interval: int or None"""
         d = datetime.datetime.now(datetime.timezone.utc)
 
         if interval is not None:
@@ -100,8 +100,7 @@ class DebugPoller:
         new one.
 
         If a timestamp file exists, returns its value.  Otherwise,
-        generate a new one with self._initial_interval.
-        """
+        generate a new one with self._initial_interval."""
         if not os.path.exists(self.TIMESTAMP_FILE) or self._force_initial_interval:
             self._logger.debug(f"Timestamp file '{self.TIMESTAMP_FILE}' doesn't exist")
             self._logger.debug(
@@ -110,14 +109,13 @@ class DebugPoller:
             return self._generate_timestamp(interval=self._initial_interval)
 
         with open(self.TIMESTAMP_FILE, "r", encoding="UTF-8") as f:
-            return f.readline()
+            return f.readline().rstrip()
 
-    def _record_timestamp(self, timestamp):
+    def record_timestamp(self, timestamp):
         """Save the timestamp into the timestamp file.
 
         :param str timestamp: Timestamp that should be saved into the
-            TIMESTAMP_FILE.
-        """
+            TIMESTAMP_FILE."""
         if self._dry_run:
             self._logger.debug("dry_run enabled, not recording timestamp file")
             return
@@ -130,6 +128,5 @@ class DebugPoller:
     def set_initial_interval(self, initial_interval):
         """Set the initial_interval value.
 
-        :param int initial_interval: The new initial_interval to be used.
-        """
+        :param int initial_interval: The new initial_interval to be used."""
         self._initial_interval = initial_interval
diff --git a/poll_launchpad.py b/poll_launchpad.py
index 161aa77..65afa28 100644
--- a/poll_launchpad.py
+++ b/poll_launchpad.py
@@ -19,16 +19,87 @@
 
 from ddebpoller import DdebPoller
 from debuginfod import grab_ddebs, grab_ddebs_sources
+from utils import DebugDB
 
-from celery import chain
+import psycopg2
 
-def poll_launchpad():
-    """Poll Launchpad."""
+
+def ensure_task(conn, cursor, msg):
+    """Return True if the task described by MSG should be processed.
+
+    Also, if True, then insert the task information into the database.
+
+    :param DebugDB conn: The database connection handler.
+
+    :param psycopg2.cursor cursor: The cursor to the SQL database.
+
+    :param dict msg: The message describing a task.
+
+    :rtype: bool, int
+    :return: True or False, and the jobid if True (or None if False)"""
+    srcname = msg["source_package"]
+    srcver = msg["version"]
+    architecture = msg["architecture"]
+    cursor.execute(
+        "SELECT * FROM tasks WHERE source = %s AND version = %s AND arch = %s",
+        (
+            srcname,
+            srcver,
+            architecture,
+        ),
+    )
+
+    if cursor.rowcount > 0:
+        return False, None
+
+    cursor.execute(
+        "INSERT INTO tasks (source, version, arch) VALUES (%s, %s, %s) RETURNING id",
+        (
+            srcname,
+            srcver,
+            architecture,
+        ),
+    )
+    jobid = cursor.fetchone()[0]
+    conn.commit()
+    return True, jobid
+
+
+def poll_launchpad(conn, cursor):
+    """Poll Launchpad and process the list of ddebs and source packages.
+
+    :param DebugDB conn: The SQL connection handler.
+
+    :param psycopg2.cursor cursor: The cursor to the SQL database."""
     poller = DdebPoller()
-    for msg in poller.get_ddebs():
-        # We have to use a chain here because the ddebs need to be
-        # downloaded before we can start fetching the sources.
-        chain(grab_ddebs.si(msg), grab_ddebs_sources.si(msg))()
+    messages, new_timestamp = poller.get_ddebs_and_sources()
+    for msg in messages:
+        proceed, jobid = ensure_task(conn, cursor, msg)
+        if proceed:
+            if jobid is None:
+                raise RuntimeError("jobid is None")
+            # We append the ID to the message so that we can efficiently
+            # remove the task once it's been processed.
+            msg["jobid"] = str(jobid)
+
+            if msg["architecture"] == "source":
+                # Source packages have their "architecture" as "source".
+                grab_ddebs_sources.delay(msg)
+            else:
+                grab_ddebs.delay(msg)
+    poller.record_timestamp(new_timestamp)
+
+
+def main():
+    # Connect to the database.
+    with DebugDB() as db:
+        with db.cursor() as cursor:
+            cursor.execute(
+                "CREATE TABLE IF NOT EXISTS tasks (id serial PRIMARY KEY, source text, version text, arch text)"
+            )
+            db.commit()
+            poll_launchpad(db, cursor)
+
 
-if __name__ == '__main__':
-    poll_launchpad()
+if __name__ == "__main__":
+    main()
diff --git a/utils.py b/utils.py
new file mode 100644
index 0000000..b577009
--- /dev/null
+++ b/utils.py
@@ -0,0 +1,48 @@
+#!/usr/bin/python3
+
+# Copyright (C) 2022 Canonical Ltd.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+# Authors: Sergio Durigan Junior <[e-mail]>
+
+import psycopg2
+
+
+class DebugDB:
+    """Abstract a connection to the SQL database."""
+
+    def __init__(self):
+        self._conn = psycopg2.connect("dbname=ubuntu-debuginfod user=mirror")
+
+    def cursor(self):
+        """Return the cursor to be used when dealing with the database."""
+        return self._conn.cursor()
+
+    def commit(self):
+        """Commit the transaction to the DB."""
+        self._conn.commit()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_traceback):
+        self._conn.commit()
+        self._conn.close()
+        self._conn = None
+
+    def __del__(self):
+        if self._conn is not None:
+            self._conn.commit()
+            self._conn.close()