feat(abi): skip objects that already have the per-minor builds they need

The verdict in abi_flat_safety describes the *flat* object, and building a
per-minor copy does not change it, so the repair list stays the same after a
run. The first fleet repair was interrupted at 14 of 18 slots having uploaded
2568 objects; resuming would have rebuilt every one of them again.

An object is repaired once a per-minor build exists for every minor that cannot
load the flat one, and the bucket is the source of truth for that. Where no
minors are recorded -- unsafe without a reason -- a build under every minor the
image carries is required, rather than assuming one is enough.

This change will:

- list the slot's objects from S3 and drop the rows already covered
- report how many were skipped, so an unexpectedly short list is visible rather
  than silent
- fail rather than proceed when the bucket cannot be listed, since an empty
  listing would make everything look unrepaired

local/list-unsafe-flat-objects.R

@@ -37,13 +37,92 @@ on.exit(try(DBI::dbDisconnect(con), silent = TRUE), add = TRUE)
 rows <- DBI::dbGetQuery(
   con,
   paste0(
-    "SELECT package, version FROM abi_flat_safety ",
+    "SELECT package, version, unsupported FROM abi_flat_safety ",
     "WHERE NOT safe AND codename = $1 AND arch = $2 ",
     "ORDER BY package, version"
   ),
   params = list(codename, arch)
 )
 
+# Drop what is already repaired. The verdict describes the *flat* object, and a
+# per-minor build does not change it, so a resumed run would otherwise rebuild
+# everything it did last time: ~2500 objects of duplicated compute after the
+# first fleet run was interrupted.
+#
+# An object is repaired once a per-minor build exists for every minor that
+# cannot load the flat one. The bucket is the source of truth for that, and it
+# cannot be routed.
+if (nrow(rows) > 0L) {
+  s3fs::s3_file_system(
+    aws_access_key_id = Sys.getenv("B2_S3_ACCESS_KEY"),
+    aws_secret_access_key = Sys.getenv("B2_S3_SECRET_KEY"),
+    endpoint = "https://s3.eu-central-003.backblazeb2.com",
+    region_name = "eu-central-003",
+    refresh = TRUE
+  )
+  contrib <- sprintf(
+    "devxy-rpkgs-binaries/%s/%s/latest/src/contrib",
+    arch,
+    codename
+  )
+  objects <- tryCatch(
+    s3fs::s3_dir_ls(contrib, recurse = TRUE),
+    error = function(e) character(0)
+  )
+  if (length(objects) == 0L) {
+    stop(
+      sprintf(
+        "Could not list %s; refusing to guess what is already repaired.",
+        contrib
+      ),
+      call. = FALSE
+    )
+  }
+  present <- sub(sprintf("^(s3://)?%s/", contrib), "", objects)
+
+  # No recorded minors means unsafe without a reason, so require a build under
+  # every minor the image carries rather than assume one is enough.
+  installed <- sub(
+    "^([0-9]+[.][0-9]+).*$",
+    "\\1",
+    basename(list.dirs("/opt/R", recursive = FALSE))
+  )
+  installed <- unique(installed[grepl("^[0-9]+[.][0-9]+$", installed)])
+
+  repaired <- vapply(
+    seq_len(nrow(rows)),
+    function(i) {
+      u <- rows$unsupported[[i]]
+      minors <- if (is.na(u) || !nzchar(u)) {
+        installed
+      } else {
+        trimws(strsplit(u, ",", fixed = TRUE)[[1L]])
+      }
+      if (length(minors) == 0L) {
+        return(FALSE)
+      }
+      wanted <- sprintf(
+        "%s/%s_%s.tar.gz",
+        minors,
+        rows$package[[i]],
+        rows$version[[i]]
+      )
+      all(wanted %in% present)
+    },
+    logical(1L)
+  )
+  if (any(repaired)) {
+    message(sprintf(
+      "%s/%s: %d of %d already have per-minor builds for every minor that needs one; skipping those.",
+      arch,
+      codename,
+      sum(repaired),
+      nrow(rows)
+    ))
+  }
+  rows <- rows[!repaired, , drop = FALSE]
+}
+
 # An empty result is ambiguous: either nothing is unsafe, or the backfill never
 # ran for this slot. Say which, rather than let the caller read silence as done.
 if (nrow(rows) == 0L) {