Fix joinRight, dockerfile optimizations, failure joining

3 files changed, 28 insertions, 23 deletions

diff --git a/worker/Dockerfile b/worker/Dockerfile
index 71e2ecf..0dfd816 100644
--- a/worker/Dockerfile
+++ b/worker/Dockerfile
@@ -1,12 +1,14 @@
 # -- <worker_dependencies> --
 FROM debian:stable-slim AS worker_dependencies
 
-ARG BITWARDEN_VERSION=2025.4.0
 RUN apt-get update && apt-get install -yqq unzip curl
 
+ARG BITWARDEN_VERSION=2025.4.0
 RUN curl -L -o /bw-linux.zip "https://github.com/bitwarden/clients/releases/download/cli-v${BITWARDEN_VERSION}/bw-linux-${BITWARDEN_VERSION}.zip"
 RUN unzip /bw-linux.zip -d / \
     && chmod +x /bw
+
+COPY --from=docker:dind /usr/local/bin/docker /usr/local/bin/
 # -- </worker_dependencies> --
 
 # -- <ci_worker> --
@@ -22,18 +24,17 @@ RUN cp -r /app/worker/jobs /var/lib/laminar/cfg
 # see: https://github.com/nodejs/docker-node/blame/89b29ef06b421598ec007605a2604ede0348b298/22/bullseye-slim/Dockerfile#L3-L4
 RUN chown -R node:node /var/lib/laminar
 
-RUN curl -fsSL https://get.docker.com | sh
-
 # adding a user to only the group"docker" doesn't deterministically give it access to the
 # docker socket of the host.
 # e.g. host has /etc/groups: docker:995, container has /etc/groups: docker:996
 # because i'm likely the only one to ever touch this, and i FORCE "docker" to be 996, this will
 # be hardcoded defaulting to 995.
 ARG DOCKER_GID="995" # but it may be overridden via this `DOCKER_GID` build arg.
-RUN groupmod -g ${DOCKER_GID} docker
-RUN usermod -a -d /var/lib/laminar -G docker  node
+RUN groupadd -g ${DOCKER_GID} docker
+RUN usermod -a -d /var/lib/laminar -G docker node
 
 COPY --from=worker_dependencies /bw /usr/local/bin/
+COPY --from=worker_dependencies /usr/local/bin/docker /usr/local/bin/
 
 USER node
 WORKDIR /var/lib/laminar
diff --git a/worker/executor.ts b/worker/executor.ts
index 11c24f6..f9fbf29 100644
--- a/worker/executor.ts
+++ b/worker/executor.ts
@@ -59,9 +59,9 @@ export const executePipeline = (
                             .flatMapAsync(TraceUtil.promiseify(TraceUtil.traceResultingEither(metric)));
                     });
                 const results = await Promise.all(parallelJobs.map((job) => job.get()));
-                const failures = results.filter((e) => e.left);
+                const failures = results.filter((e) => e.left().present());
                 if (failures.length > 0) {
-                    return Either.left(new Error(failures.toString()));
+                    return Either.left(new Error(JSON.stringify(failures)));
                 }
             }
             return Either.right(<void>undefined);
diff --git a/worker/scripts/checkout_ci.ts b/worker/scripts/checkout_ci.ts
index 6a1dcad..ac36d69 100755
--- a/worker/scripts/checkout_ci.ts
+++ b/worker/scripts/checkout_ci.ts
@@ -11,10 +11,11 @@ import {
     prependWith,
     TraceUtil,
     getStdoutMany,
+    IEither,
 } from '@emprespresso/pengueno';
 import { mkdir, readFile, rm } from 'fs/promises';
 import { join } from 'path';
-import { type CheckoutCiJob, type FetchCodeJob, PipelineImpl } from '@emprespresso/ci_model';
+import { type CheckoutCiJob, type FetchCodeJob, Pipeline, PipelineImpl } from '@emprespresso/ci_model';
 import { executeJob, executePipeline } from '@emprespresso/ci_worker';
 
 interface CiWorkflow {
@@ -102,21 +103,24 @@ await LogMetricTraceable.ofLogTraceable(logTraceableJob)
                 commands: getPipelineGenerationCommand(job, workflow),
             }));
     })
-    .map(async (tEitherPipelineGenerationCommand) => {
-        const eitherJobCommand = await tEitherPipelineGenerationCommand.get();
-        const pipelineSerialized = await eitherJobCommand.flatMapAsync(({ commands }) =>
-            getStdoutMany(
-                tEitherPipelineGenerationCommand.move(commands).peek((t) => t.trace.trace(JSON.stringify(t.get()))),
-            ),
-        );
-        return pipelineSerialized
-            .flatMap((results) => {
-                tEitherPipelineGenerationCommand.trace.trace(JSON.stringify(results));
-                const pipeline = results.at(-1)!;
-                return PipelineImpl.from(pipeline);
-            })
-            .joinRight(eitherJobCommand, (job, pipeline) => ({ job, pipeline }));
-    })
+    .map(
+        async (
+            tEitherPipelineGenerationCommand,
+        ): Promise<IEither<Error, { job: CheckoutCiJob; pipeline: Pipeline }>> => {
+            const eitherJobCommand = await tEitherPipelineGenerationCommand.get();
+            const pipelineSerialized = await eitherJobCommand.flatMapAsync(({ commands }) =>
+                getStdoutMany(
+                    tEitherPipelineGenerationCommand.move(commands).peek((t) => t.trace.trace(JSON.stringify(t.get()))),
+                ),
+            );
+            return pipelineSerialized
+                .flatMap((results) => {
+                    const pipeline = results.at(-1)!;
+                    return PipelineImpl.from(pipeline);
+                })
+                .joinRight(eitherJobCommand, ({ job }, pipeline) => ({ job, pipeline }));
+        },
+    )
     .peek(
         TraceUtil.promiseify((tEitherPipeline) =>
             tEitherPipeline