From 09a82ea73b90bb4469994ed02c5c7c778ce90226 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Mon, 31 Aug 2026 14:19:48 +0200 Subject: [PATCH 01/15] ci: route workflows to R730 runners --- .github/workflows/ci.yml | 2 +- .github/workflows/coverage.yml | 2 +- .github/workflows/docker.yml | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f0e65a0..4a943f5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,7 +15,7 @@ concurrency: jobs: go: - runs-on: ${{ github.event_name == 'pull_request' && 'ubuntu-24.04' || fromJSON('["self-hosted","Linux","X64","arko","downloader"]') }} + runs-on: ${{ github.event_name == 'pull_request' && 'ubuntu-24.04' || fromJSON('["self-hosted","Linux","X64","r730","downloader"]') }} permissions: contents: read steps: diff --git a/.github/workflows/coverage.yml b/.github/workflows/coverage.yml index 4e0297f..783f37c 100644 --- a/.github/workflows/coverage.yml +++ b/.github/workflows/coverage.yml @@ -15,7 +15,7 @@ concurrency: jobs: coverage: - runs-on: ${{ github.event_name == 'pull_request' && 'ubuntu-24.04' || fromJSON('["self-hosted","Linux","X64","arko","downloader"]') }} + runs-on: ${{ github.event_name == 'pull_request' && 'ubuntu-24.04' || fromJSON('["self-hosted","Linux","X64","r730","downloader"]') }} permissions: contents: read steps: diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index f3d53d8..63642aa 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -20,7 +20,7 @@ concurrency: jobs: build-and-push: - runs-on: [self-hosted, Linux, X64, arko, downloader] + runs-on: [self-hosted, Linux, X64, r730, downloader] permissions: contents: read packages: write @@ -141,7 +141,7 @@ jobs: notify-orchestrator: needs: build-and-push if: github.ref_name == 'dev' || github.ref_name == 'main' || startsWith(github.ref, 'refs/tags/v') - runs-on: [self-hosted, Linux, X64, arko, downloader] + runs-on: [self-hosted, Linux, X64, r730, downloader] permissions: contents: read env: From f02d7b547ad1046f725c4daffab12febae226888 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Mon, 31 Aug 2026 15:21:29 +0200 Subject: [PATCH 02/15] perf: parallelize multi-architecture image builds --- .github/actionlint.yaml | 1 + .github/workflows/docker.yml | 158 ++++++++++++++++++++++++++++------- 2 files changed, 128 insertions(+), 31 deletions(-) diff --git a/.github/actionlint.yaml b/.github/actionlint.yaml index 927074b..ce09d65 100644 --- a/.github/actionlint.yaml +++ b/.github/actionlint.yaml @@ -1,4 +1,5 @@ self-hosted-runner: labels: - arko + - r730 - downloader diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index 63642aa..fb4782a 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -19,23 +19,20 @@ concurrency: cancel-in-progress: true jobs: - build-and-push: + prepare: runs-on: [self-hosted, Linux, X64, r730, downloader] permissions: contents: read - packages: write outputs: - digest: ${{ steps.build.outputs.digest }} + build-time: ${{ steps.build-info.outputs.build-time }} image: ${{ steps.build-info.outputs.image }} + labels: ${{ steps.meta.outputs.labels }} + metadata-json: ${{ steps.meta.outputs.json }} version: ${{ steps.build-info.outputs.version }} - steps: - name: Checkout uses: actions/checkout@v7 - - name: Isolate Docker credentials - run: echo "DOCKER_CONFIG=$RUNNER_TEMP/docker-config" >> "$GITHUB_ENV" - - name: Resolve build metadata id: build-info env: @@ -81,13 +78,6 @@ jobs: echo "build-time=$(date -u +'%Y-%m-%dT%H:%M:%SZ')" } >> "$GITHUB_OUTPUT" - - name: Log in to GitHub Container Registry - uses: docker/login-action@v4 - with: - registry: ${{ env.REGISTRY }} - username: ${{ github.actor }} - password: ${{ secrets.GITHUB_TOKEN }} - - name: Extract metadata id: meta uses: docker/metadata-action@v6 @@ -104,7 +94,37 @@ jobs: type=raw,value=latest,enable=${{ github.ref_name == 'dev' }} type=raw,value=beta,enable=${{ github.ref_name == 'dev' }} + build-platform: + needs: prepare + runs-on: [self-hosted, Linux, X64, r730, downloader] + timeout-minutes: 30 + permissions: + contents: read + packages: write + strategy: + fail-fast: false + matrix: + include: + - platform: linux/amd64 + arch: amd64 + - platform: linux/arm64 + arch: arm64 + steps: + - name: Checkout + uses: actions/checkout@v7 + + - name: Isolate Docker credentials + run: echo "DOCKER_CONFIG=$RUNNER_TEMP/docker-config" >> "$GITHUB_ENV" + + - name: Log in to GitHub Container Registry + uses: docker/login-action@v4 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + - name: Set up QEMU + if: matrix.arch == 'arm64' uses: docker/setup-qemu-action@v4 with: platforms: arm64 @@ -113,33 +133,109 @@ jobs: - name: Set up Docker Buildx uses: docker/setup-buildx-action@v4 - - name: Build and push + - name: Build and push platform image id: build uses: docker/build-push-action@v7 with: context: . file: Dockerfile.wolfi build-args: | - BUILD_VERSION=${{ steps.build-info.outputs.version }} + BUILD_VERSION=${{ needs.prepare.outputs.version }} BUILD_REVISION=${{ github.sha }} - BUILD_TIME=${{ steps.build-info.outputs.build-time }} - platforms: linux/amd64,linux/arm64 - push: true - tags: ${{ steps.meta.outputs.tags }} - labels: ${{ steps.meta.outputs.labels }} - cache-from: type=gha - cache-to: type=gha,mode=max + BUILD_TIME=${{ needs.prepare.outputs.build-time }} + platforms: ${{ matrix.platform }} + labels: ${{ needs.prepare.outputs.labels }} + outputs: type=image,name=${{ needs.prepare.outputs.image }},push-by-digest=true,name-canonical=true,push=true + provenance: false + cache-from: type=gha,scope=downloader-${{ matrix.arch }} + cache-to: type=gha,mode=max,scope=downloader-${{ matrix.arch }} - name: Verify published FFmpeg runtime env: - IMAGE: ${{ steps.build-info.outputs.image }} + IMAGE: ${{ needs.prepare.outputs.image }} + DIGEST: ${{ steps.build.outputs.digest }} + run: | + docker pull "$IMAGE@$DIGEST" + docker run --rm --entrypoint ffmpeg "$IMAGE@$DIGEST" -version + + - name: Export digest + env: + DIGEST: ${{ steps.build.outputs.digest }} run: | - image="$IMAGE:sha-${GITHUB_SHA::7}" - docker pull "$image" - docker run --rm --entrypoint ffmpeg "$image" -version + digest_dir="$RUNNER_TEMP/downloader-digests" + rm -rf "$digest_dir" + mkdir -p "$digest_dir" + touch "$digest_dir/${DIGEST#sha256:}" + + - name: Upload digest + uses: actions/upload-artifact@v7 + with: + name: downloader-digest-${{ matrix.arch }}-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/downloader-digests/* + if-no-files-found: error + retention-days: 1 + + publish: + needs: [prepare, build-platform] + runs-on: [self-hosted, Linux, X64, r730, downloader] + timeout-minutes: 10 + permissions: + contents: read + packages: write + outputs: + digest: ${{ steps.manifest.outputs.digest }} + steps: + - name: Isolate Docker credentials + run: echo "DOCKER_CONFIG=$RUNNER_TEMP/docker-config" >> "$GITHUB_ENV" + + - name: Prepare digest directory + run: rm -rf "$RUNNER_TEMP/downloader-digests" + + - name: Download digests + uses: actions/download-artifact@v8 + with: + path: ${{ runner.temp }}/downloader-digests + pattern: downloader-digest-*-${{ github.run_id }}-${{ github.run_attempt }} + merge-multiple: true + + - name: Log in to GitHub Container Registry + uses: docker/login-action@v4 + with: + registry: ${{ env.REGISTRY }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v4 + + - name: Publish manifest list + id: manifest + env: + DIGEST_DIR: ${{ runner.temp }}/downloader-digests + IMAGE: ${{ needs.prepare.outputs.image }} + METADATA_JSON: ${{ needs.prepare.outputs.metadata-json }} + run: | + mapfile -t digests < <(find "$DIGEST_DIR" -maxdepth 1 -type f -printf '%f\n' | sort) + if [[ "${#digests[@]}" -ne 2 ]]; then + echo "Expected two platform digests, found ${#digests[@]}" + exit 1 + fi + mapfile -t tags < <(jq -r '.tags[]' <<< "$METADATA_JSON") + tag_args=() + for tag in "${tags[@]}"; do tag_args+=(--tag "$tag"); done + source_args=() + for digest in "${digests[@]}"; do source_args+=("$IMAGE@sha256:$digest"); done + docker buildx imagetools create "${tag_args[@]}" "${source_args[@]}" + manifest_json="$(docker buildx imagetools inspect "${tags[0]}" --format '{{json .Manifest}}')" + digest="$(jq -r '.digest' <<< "$manifest_json")" + if [[ "$digest" != sha256:* ]]; then + echo "Published manifest has no valid digest" + exit 1 + fi + echo "digest=$digest" >> "$GITHUB_OUTPUT" notify-orchestrator: - needs: build-and-push + needs: [prepare, publish] if: github.ref_name == 'dev' || github.ref_name == 'main' || startsWith(github.ref, 'refs/tags/v') runs-on: [self-hosted, Linux, X64, r730, downloader] permissions: @@ -149,9 +245,9 @@ jobs: steps: - name: Send image digest env: - DIGEST: ${{ needs.build-and-push.outputs.digest }} - IMAGE: ${{ needs.build-and-push.outputs.image }} - VERSION: ${{ needs.build-and-push.outputs.version }} + DIGEST: ${{ needs.publish.outputs.digest }} + IMAGE: ${{ needs.prepare.outputs.image }} + VERSION: ${{ needs.prepare.outputs.version }} run: | if [[ "$GITHUB_REF_NAME" == "dev" ]]; then channel="beta"; else channel="stable"; fi jq -n \ From a156c5cae859d5a1d06762589267454bebc28540 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Mon, 31 Aug 2026 20:21:43 +0200 Subject: [PATCH 03/15] fix: verify downloader image on target platform --- .github/workflows/docker.yml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index fb4782a..87a6d9b 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -154,9 +154,10 @@ jobs: env: IMAGE: ${{ needs.prepare.outputs.image }} DIGEST: ${{ steps.build.outputs.digest }} + PLATFORM: ${{ matrix.platform }} run: | - docker pull "$IMAGE@$DIGEST" - docker run --rm --entrypoint ffmpeg "$IMAGE@$DIGEST" -version + docker pull --platform "$PLATFORM" "$IMAGE@$DIGEST" + docker run --rm --platform "$PLATFORM" --entrypoint ffmpeg "$IMAGE@$DIGEST" -version - name: Export digest env: From b8a20724fe9fe92c4c85fcf91138fef4ffbd5f1d Mon Sep 17 00:00:00 2001 From: Priveetee Date: Mon, 31 Aug 2026 20:36:46 +0200 Subject: [PATCH 04/15] fix: validate downloader arm image without emulation --- .github/workflows/docker.yml | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index 87a6d9b..3e17132 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -155,9 +155,18 @@ jobs: IMAGE: ${{ needs.prepare.outputs.image }} DIGEST: ${{ steps.build.outputs.digest }} PLATFORM: ${{ matrix.platform }} + ARCH: ${{ matrix.arch }} run: | docker pull --platform "$PLATFORM" "$IMAGE@$DIGEST" - docker run --rm --platform "$PLATFORM" --entrypoint ffmpeg "$IMAGE@$DIGEST" -version + if [[ "$ARCH" == "amd64" ]]; then + docker run --rm --platform "$PLATFORM" --entrypoint ffmpeg "$IMAGE@$DIGEST" -version + else + container_id="$(docker create --platform "$PLATFORM" --entrypoint ffmpeg "$IMAGE@$DIGEST")" + trap 'docker rm "$container_id" >/dev/null' EXIT + ffmpeg_path="$RUNNER_TEMP/ffmpeg-$ARCH" + docker cp "$container_id:/usr/bin/ffmpeg" "$ffmpeg_path" + readelf -h "$ffmpeg_path" | grep -q 'Machine:.*AArch64' + fi - name: Export digest env: From 2120ae26472afb72049d05a8e1da27040dc984bc Mon Sep 17 00:00:00 2001 From: Tax_Tux <138765817+Priveetee@users.noreply.github.com> Date: Mon, 31 Aug 2026 21:31:05 +0200 Subject: [PATCH 05/15] chore: benchmark dev runners From c50a6dc0955114cee0767f9065bf99e2d8599385 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Tue, 8 Sep 2026 11:40:28 +0200 Subject: [PATCH 06/15] chore: bump downloader version to 1.8.1 --- VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VERSION b/VERSION index bd8bf88..a8fdfda 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.7.0 +1.8.1 From d945e5f19fffce715e5931fb93d56d32b0cab9a3 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 09:05:18 +0200 Subject: [PATCH 07/15] fix: preserve job JSON with PostgreSQL simple protocol --- internal/db/postgres.go | 2 +- internal/db/postgres_integration_test.go | 105 +++++++++++++++++++++++ 2 files changed, 106 insertions(+), 1 deletion(-) create mode 100644 internal/db/postgres_integration_test.go diff --git a/internal/db/postgres.go b/internal/db/postgres.go index 3231b7e..78ff6c4 100644 --- a/internal/db/postgres.go +++ b/internal/db/postgres.go @@ -71,7 +71,7 @@ func (s *PostgresSink) upsert(ctx context.Context, record *job.Record) error { } _, err = s.pool.Exec(ctx, upsertSQL, record.ID, record.CacheKey, record.URL, string(record.Status), record.Title, - options, progress, resolved, record.Artifact, record.Storage, + string(options), string(progress), string(resolved), record.Artifact, record.Storage, record.ExpiresAt, stringPtr(record.Error), stringPtr(record.ErrorCode), record.QueuedAt, record.StartedAt, record.FinishedAt, record.DownloadMs, record.MuxMs, record.TotalMs, ) diff --git a/internal/db/postgres_integration_test.go b/internal/db/postgres_integration_test.go new file mode 100644 index 0000000..3f95bda --- /dev/null +++ b/internal/db/postgres_integration_test.go @@ -0,0 +1,105 @@ +package db + +import ( + "context" + "fmt" + "os" + "reflect" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "typetype-downloader-go/internal/job" +) + +func TestPostgresJobRoundTrip(t *testing.T) { + url := os.Getenv("TYPETYPE_TEST_POSTGRES_URL") + if url == "" { + t.Skip("set TYPETYPE_TEST_POSTGRES_URL to an isolated test database") + } + for _, mode := range []pgx.QueryExecMode{pgx.QueryExecModeCacheStatement, pgx.QueryExecModeSimpleProtocol} { + t.Run(mode.String(), func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + config, err := pgxpool.ParseConfig(url) + if err != nil { + t.Fatal(err) + } + admin, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + t.Fatal(err) + } + defer admin.Close() + name := fmt.Sprintf("downloader_test_%d", time.Now().UnixNano()) + schemaName := pgx.Identifier{name}.Sanitize() + if _, err := admin.Exec(ctx, "create schema "+schemaName); err != nil { + t.Fatal(err) + } + defer func() { + cleanup, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + if _, err := admin.Exec(cleanup, "drop schema "+schemaName+" cascade"); err != nil { + t.Error(err) + } + }() + config = config.Copy() + config.ConnConfig.RuntimeParams["search_path"] = name + config.ConnConfig.DefaultQueryExecMode = mode + pool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + if err := migrate(ctx, pool); err != nil { + t.Fatal(err) + } + sink := &PostgresSink{pool: pool} + now := time.Now().UTC().Truncate(time.Microsecond) + expires := now.Add(time.Hour) + record := &job.Record{ + ID: "round-trip", CacheKey: "cache", URL: "https://example.com/video", + Status: job.StatusQueued, Title: "Sample", QueuedAt: now, + Options: job.Options{Mode: "video", Quality: "720p"}, + Progress: job.Progress{Stage: "queued"}, + } + if err := sink.upsert(ctx, record); err != nil { + t.Fatal(err) + } + queued, err := sink.LoadRunnable(ctx) + if err != nil || len(queued) != 1 { + t.Fatalf("queued jobs: %v, %v", queued, err) + } + if queued[0].Status != job.StatusQueued || !queued[0].QueuedAt.Equal(now) || + !reflect.DeepEqual(queued[0].Options, record.Options) || queued[0].FinishedAt != nil { + t.Fatalf("unexpected queued record: %+v", queued[0]) + } + record.Status = job.StatusDone + record.Artifact = "sample.mp4" + record.Storage = "s3" + record.FinishedAt = &now + record.ExpiresAt = &expires + record.Resolved = &job.ResolvedOutput{Container: "mp4", Height: 720} + record.Progress = job.Progress{Stage: "done", DownloadedBytes: 12345, TotalBytes: 12345} + if err := sink.upsert(ctx, record); err != nil { + t.Fatal(err) + } + done, err := sink.LoadDone(ctx) + if err != nil || len(done) != 1 { + t.Fatalf("done jobs: %v, %v", done, err) + } + got := done[0] + if got.Status != job.StatusDone || got.Artifact != record.Artifact || got.Storage != "s3" || + got.FinishedAt == nil || !got.FinishedAt.Equal(now) || + got.ExpiresAt == nil || !got.ExpiresAt.Equal(expires) || + !reflect.DeepEqual(got.Resolved, record.Resolved) || got.Progress != record.Progress { + t.Fatalf("unexpected completed record: %+v", got) + } + queued, err = sink.LoadRunnable(ctx) + if err != nil || len(queued) != 0 { + t.Fatalf("completed job remains runnable: %v, %v", queued, err) + } + }) + } +} From c5edbd9130ed31a01322b2ad317233e790a3f167 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 09:05:18 +0200 Subject: [PATCH 08/15] test: verify S3 artifact upload content and deletion --- internal/artifact/s3_upload_test.go | 88 +++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 internal/artifact/s3_upload_test.go diff --git a/internal/artifact/s3_upload_test.go b/internal/artifact/s3_upload_test.go new file mode 100644 index 0000000..42a6384 --- /dev/null +++ b/internal/artifact/s3_upload_test.go @@ -0,0 +1,88 @@ +package artifact + +import ( + "bytes" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/minio/minio-go/v7" + "github.com/minio/minio-go/v7/pkg/credentials" +) + +func TestS3StoreUploadsExactFileAndDeletesArtifact(t *testing.T) { + payload := bytes.Repeat([]byte("artifact-content"), 8192) + uploads := make(chan []byte, 1) + deletes := make(chan struct{}, 1) + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/downloads/artifact.mp4" { + t.Errorf("unexpected object path: %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + if !strings.HasPrefix(r.Header.Get("Authorization"), "AWS4-HMAC-SHA256 ") { + t.Error("missing signed request") + } + switch r.Method { + case http.MethodPut: + uploaded, err := io.ReadAll(io.LimitReader(r.Body, int64(len(payload)+1))) + if err != nil { + t.Error(err) + } + if r.Header.Get("Content-Type") != "video/mp4" { + t.Errorf("content type: %s", r.Header.Get("Content-Type")) + } + w.Header().Set("ETag", `"test-artifact"`) + uploads <- uploaded + case http.MethodDelete: + deletes <- struct{}{} + w.WriteHeader(http.StatusNoContent) + default: + t.Errorf("unexpected method: %s", r.Method) + w.WriteHeader(http.StatusMethodNotAllowed) + } + })) + defer server.Close() + endpoint := strings.TrimPrefix(server.URL, "https://") + store, err := NewS3Store(S3Config{ + Endpoint: endpoint, Region: "test", Bucket: "downloads", + AccessKey: "test-key", SecretKey: "test-secret", UseSSL: true, PathStyle: true, + URLTTL: time.Hour, + }) + if err != nil { + t.Fatal(err) + } + store.client, err = minio.New(endpoint, &minio.Options{ + Creds: credentials.NewStaticV4("test-key", "test-secret", ""), + Secure: true, Region: "test", BucketLookup: minio.BucketLookupPath, + Transport: server.Client().Transport, + }) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "artifact.mp4") + if err := os.WriteFile(path, payload, 0o600); err != nil { + t.Fatal(err) + } + saved, err := store.Save(t.Context(), path, "artifact.mp4") + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(<-uploads, payload) || saved.Backend != "s3" || saved.Location != "artifact.mp4" || + !saved.Expires.After(time.Now()) { + t.Fatal("artifact bytes or metadata changed during upload") + } + if err := store.Delete(t.Context(), saved); err != nil { + t.Fatal(err) + } + select { + case <-deletes: + default: + t.Fatal("artifact was not deleted") + } +} From 3bdbda36d54442c4099c98e793afb391069fe33d Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 09:05:18 +0200 Subject: [PATCH 09/15] chore: update PostgreSQL and S3 client dependencies --- go.mod | 19 +++++++++---------- go.sum | 39 ++++++++++++++++----------------------- 2 files changed, 25 insertions(+), 33 deletions(-) diff --git a/go.mod b/go.mod index a7ab7c4..54c2c74 100644 --- a/go.mod +++ b/go.mod @@ -1,12 +1,12 @@ module typetype-downloader-go -go 1.26 +go 1.26.0 require ( - github.com/jackc/pgx/v5 v5.10.0 - github.com/minio/minio-go/v7 v7.2.1 + github.com/jackc/pgx/v5 v5.11.0 + github.com/minio/minio-go/v7 v7.3.0 github.com/redis/go-redis/v9 v9.22.0 - golang.org/x/sys v0.47.0 + golang.org/x/sys v0.48.0 ) require ( @@ -16,10 +16,9 @@ require ( github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect - github.com/klauspost/compress v1.19.0 // indirect + github.com/klauspost/compress v1.19.2 // indirect github.com/klauspost/cpuid/v2 v2.4.0 // indirect github.com/klauspost/crc32 v1.3.0 // indirect - github.com/kr/text v0.2.0 // indirect github.com/minio/crc64nvme v1.1.1 // indirect github.com/minio/md5-simd v1.1.2 // indirect github.com/philhofer/fwd v1.2.0 // indirect @@ -27,10 +26,10 @@ require ( github.com/tinylib/msgp v1.6.4 // indirect github.com/zeebo/xxh3 v1.1.0 // indirect go.uber.org/atomic v1.11.0 // indirect - go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/crypto v0.54.0 // indirect - golang.org/x/net v0.57.0 // indirect + go.yaml.in/yaml/v3 v3.0.5 // indirect + golang.org/x/crypto v0.55.0 // indirect + golang.org/x/net v0.58.0 // indirect golang.org/x/sync v0.22.0 // indirect - golang.org/x/text v0.40.0 // indirect + golang.org/x/text v0.41.0 // indirect gopkg.in/ini.v1 v1.67.3 // indirect ) diff --git a/go.sum b/go.sum index efc89d7..249699c 100644 --- a/go.sum +++ b/go.sum @@ -4,7 +4,6 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -18,33 +17,29 @@ github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7Ulw github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg= +github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= -github.com/klauspost/compress v1.19.0 h1:sXLILfc9jV2QYWkzFOPWStmcUVH2RHEB1JCdY2oVvCQ= -github.com/klauspost/compress v1.19.0/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8= +github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.4.0 h1:S6Hrbc7+ywsr0r+RLapfGBHfyefhCTwEh3A0tV913Dw= github.com/klauspost/cpuid/v2 v2.4.0/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU= github.com/klauspost/crc32 v1.3.0 h1:sSmTt3gUt81RP655XGZPElI0PelVTZ6YwCRnPSupoFM= github.com/klauspost/crc32 v1.3.0/go.mod h1:D7kQaZhnkX/Y0tstFGf8VUzv2UofNGqCjnC3zdHB0Hw= -github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= -github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= -github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= -github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/minio/crc64nvme v1.1.1 h1:8dwx/Pz49suywbO+auHCBpCtlW1OfpcLN7wYgVR6wAI= github.com/minio/crc64nvme v1.1.1/go.mod h1:eVfm2fAzLlxMdUGc0EEBGSMmPwmXD5XiNRpnu9J3bvg= github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= -github.com/minio/minio-go/v7 v7.2.1 h1:PfBfwvKB/MmqyN8Vb1G9voWisaM9OrLv+WwOvMwS9Dw= -github.com/minio/minio-go/v7 v7.2.1/go.mod h1:EU9hENAStx/xXduNdrGO5e4X5vk19NtgB+RIPjZO8o0= +github.com/minio/minio-go/v7 v7.3.0 h1:HM4pFCSQq/TK+j0/zmorSh5ddh81iDgRgU0BG0Vz/YU= +github.com/minio/minio-go/v7 v7.3.0/go.mod h1:KUPWdecEO1LWyUz+sTGXAuf2jZHrPh5fCsRH86QbPfk= github.com/philhofer/fwd v1.2.0 h1:e6DnBTl7vGY+Gz322/ASL4Gyp1FspeMvx1RNDoToZuM= github.com/philhofer/fwd v1.2.0/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= -github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= -github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU= github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= @@ -66,21 +61,19 @@ github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= -go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= -go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= -golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= -golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= -golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE= -golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU= +go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= +go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= +golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= +golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= -golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= -golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= -golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= -golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/ini.v1 v1.67.3 h1:iM9Lhz5MRSGhHVGGwCuzG9KO8PoirCXj/m/qTmOJJQw= gopkg.in/ini.v1 v1.67.3/go.mod h1:x/cyOwCgZqOkJoDIJ3c1KNHMo10+nLGAhh+kn3Zizss= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= From 3b4d856cb38c6943ae09ee1c1a5ea1da382910f2 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 09:13:28 +0200 Subject: [PATCH 10/15] fix: drain and stop persistence workers on shutdown --- internal/db/dragonfly.go | 30 ++++---- internal/db/job_writer.go | 91 ++++++++++++++++++++++ internal/db/job_writer_test.go | 134 +++++++++++++++++++++++++++++++++ internal/db/postgres.go | 29 +++---- 4 files changed, 250 insertions(+), 34 deletions(-) create mode 100644 internal/db/job_writer.go create mode 100644 internal/db/job_writer_test.go diff --git a/internal/db/dragonfly.go b/internal/db/dragonfly.go index 831462b..3fd1ecc 100644 --- a/internal/db/dragonfly.go +++ b/internal/db/dragonfly.go @@ -3,6 +3,8 @@ package db import ( "context" "encoding/json" + "errors" + "log/slog" "time" "github.com/redis/go-redis/v9" @@ -13,7 +15,7 @@ import ( type DragonflySink struct { client *redis.Client ttl time.Duration - queue chan *job.Record + writer *jobWriter } func OpenDragonfly(addr string, ttlSeconds int) *DragonflySink { @@ -21,33 +23,27 @@ func OpenDragonfly(addr string, ttlSeconds int) *DragonflySink { ttlSeconds = 600 } sink := &DragonflySink{ - client: redis.NewClient(&redis.Options{Addr: addr}), + client: redis.NewClient(&redis.Options{Addr: addr, ContextTimeoutEnabled: true}), ttl: time.Duration(ttlSeconds) * time.Second, - queue: make(chan *job.Record, 2048), } - go sink.run() + sink.writer = newJobWriter(sink.save, time.Second, persistenceShutdownTimeout) return sink } -func (s *DragonflySink) Close() error { return s.client.Close() } +func (s *DragonflySink) Close() error { + err := s.writer.close() + if err != nil { + slog.Error("dragonfly persistence shutdown", "error", err) + } + return errors.Join(err, s.client.Close()) +} func (s *DragonflySink) Name() string { return "dragonfly" } func (s *DragonflySink) Health(ctx context.Context) error { return s.client.Ping(ctx).Err() } func (s *DragonflySink) SaveJob(record *job.Record) { - select { - case s.queue <- record: - default: - } -} - -func (s *DragonflySink) run() { - for record := range s.queue { - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - _ = s.save(ctx, record) - cancel() - } + s.writer.enqueue(record) } func (s *DragonflySink) save(ctx context.Context, record *job.Record) error { diff --git a/internal/db/job_writer.go b/internal/db/job_writer.go new file mode 100644 index 0000000..52652e7 --- /dev/null +++ b/internal/db/job_writer.go @@ -0,0 +1,91 @@ +package db + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "time" + + "typetype-downloader-go/internal/job" +) + +const persistenceShutdownTimeout = 5 * time.Second + +type jobWriter struct { + queue chan *job.Record + done chan struct{} + ctx context.Context + cancel context.CancelFunc + save func(context.Context, *job.Record) error + writeTimeout time.Duration + shutdownTimeout time.Duration + mu sync.RWMutex + closed bool + closeOnce sync.Once + rejected atomic.Int64 + failures int + lastError error +} + +func newJobWriter(save func(context.Context, *job.Record) error, writeTimeout, shutdownTimeout time.Duration) *jobWriter { + ctx, cancel := context.WithCancel(context.Background()) + writer := &jobWriter{ + queue: make(chan *job.Record, 2048), done: make(chan struct{}), + ctx: ctx, cancel: cancel, save: save, + writeTimeout: writeTimeout, shutdownTimeout: shutdownTimeout, + } + go writer.run() + return writer +} + +func (w *jobWriter) enqueue(record *job.Record) { + w.mu.RLock() + defer w.mu.RUnlock() + if w.closed { + return + } + select { + case w.queue <- record: + default: + w.rejected.Add(1) + } +} + +func (w *jobWriter) run() { + defer close(w.done) + for record := range w.queue { + if err := w.ctx.Err(); err != nil { + w.failures++ + w.lastError = err + continue + } + ctx, cancel := context.WithTimeout(w.ctx, w.writeTimeout) + err := w.save(ctx, record) + cancel() + if err != nil { + w.failures++ + w.lastError = err + } + } +} + +func (w *jobWriter) close() error { + w.closeOnce.Do(func() { + w.mu.Lock() + w.closed = true + close(w.queue) + w.mu.Unlock() + timer := time.AfterFunc(w.shutdownTimeout, w.cancel) + <-w.done + timer.Stop() + w.cancel() + }) + if w.failures > 0 { + return fmt.Errorf("persistence: %d writes failed, %d rejected: %w", w.failures, w.rejected.Load(), w.lastError) + } + if rejected := w.rejected.Load(); rejected > 0 { + return fmt.Errorf("persistence queue rejected %d writes", rejected) + } + return nil +} diff --git a/internal/db/job_writer_test.go b/internal/db/job_writer_test.go new file mode 100644 index 0000000..732e65e --- /dev/null +++ b/internal/db/job_writer_test.go @@ -0,0 +1,134 @@ +package db + +import ( + "context" + "errors" + "reflect" + "strings" + "sync" + "testing" + "time" + + "typetype-downloader-go/internal/job" +) + +func TestJobWriterDrainsInOrder(t *testing.T) { + var saved []*job.Record + w := newJobWriter(func(_ context.Context, r *job.Record) error { + saved = append(saved, r) + return nil + }, time.Second, time.Second) + want := []*job.Record{{}, {}, {}} + for _, r := range want { + w.enqueue(r) + } + if err := w.close(); err != nil { + t.Fatal(err) + } + w.enqueue(&job.Record{}) + if err := w.close(); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(saved, want) || len(saved) != 3 { + t.Fatalf("saved %d records, want 3 in order", len(saved)) + } + for i := range want { + if saved[i] != want[i] { + t.Fatal("record order changed") + } + } + if len(w.queue) != 0 || w.ctx.Err() == nil { + t.Fatal("writer retained queued records or an active context") + } +} + +func TestJobWriterReportsFailureAndContinues(t *testing.T) { + want := errors.New("write failed") + calls := 0 + w := newJobWriter(func(context.Context, *job.Record) error { + calls++ + if calls == 1 { + return want + } + return nil + }, time.Second, time.Second) + w.enqueue(&job.Record{}) + w.enqueue(&job.Record{}) + if err := w.close(); !errors.Is(err, want) || calls != 2 { + t.Fatalf("close=%v calls=%d", err, calls) + } +} + +func TestJobWriterWriteDeadline(t *testing.T) { + finished := make(chan struct{}) + w := newJobWriter(func(ctx context.Context, _ *job.Record) error { + <-ctx.Done() + close(finished) + return ctx.Err() + }, 10*time.Millisecond, time.Second) + w.enqueue(&job.Record{}) + select { + case <-finished: + case <-time.After(time.Second): + w.cancel() + _ = w.close() + t.Fatal("write deadline was not propagated") + } + if err := w.close(); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("close=%v", err) + } +} + +func TestJobWriterShutdownCancelsPendingWrites(t *testing.T) { + started := make(chan struct{}) + w := newJobWriter(func(ctx context.Context, _ *job.Record) error { + close(started) + <-ctx.Done() + return ctx.Err() + }, time.Hour, 20*time.Millisecond) + w.enqueue(&job.Record{}) + <-started + w.enqueue(&job.Record{}) + if err := w.close(); !errors.Is(err, context.Canceled) { + t.Fatalf("close=%v", err) + } + if w.failures != 2 || len(w.queue) != 0 { + t.Fatalf("failures=%d pending=%d", w.failures, len(w.queue)) + } +} + +func TestJobWriterReportsFullQueue(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + w := newJobWriter(func(context.Context, *job.Record) error { + once.Do(func() { close(started); <-release }) + return nil + }, time.Second, time.Second) + w.enqueue(&job.Record{}) + <-started + for i := 0; i <= cap(w.queue); i++ { + w.enqueue(&job.Record{}) + } + close(release) + if err := w.close(); err == nil || !strings.Contains(err.Error(), "rejected 1 writes") { + t.Fatalf("close=%v", err) + } +} + +func TestJobWriterConcurrentCloseAndEnqueue(t *testing.T) { + w := newJobWriter(func(context.Context, *job.Record) error { return nil }, time.Second, time.Second) + var wg sync.WaitGroup + for i := 0; i < 16; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 50; j++ { + w.enqueue(&job.Record{}) + } + if err := w.close(); err != nil { + t.Error(err) + } + }() + } + wg.Wait() +} diff --git a/internal/db/postgres.go b/internal/db/postgres.go index 78ff6c4..f7ae60a 100644 --- a/internal/db/postgres.go +++ b/internal/db/postgres.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "log/slog" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -12,8 +13,8 @@ import ( ) type PostgresSink struct { - pool *pgxpool.Pool - queue chan *job.Record + pool *pgxpool.Pool + writer *jobWriter } func OpenPostgres(ctx context.Context, databaseURL string) (*PostgresSink, error) { @@ -25,30 +26,24 @@ func OpenPostgres(ctx context.Context, databaseURL string) (*PostgresSink, error pool.Close() return nil, err } - sink := &PostgresSink{pool: pool, queue: make(chan *job.Record, 2048)} - go sink.run() + sink := &PostgresSink{pool: pool} + sink.writer = newJobWriter(sink.upsert, 5*time.Second, persistenceShutdownTimeout) return sink, nil } -func (s *PostgresSink) Close() { s.pool.Close() } +func (s *PostgresSink) Close() { + if err := s.writer.close(); err != nil { + slog.Error("postgres persistence shutdown", "error", err) + } + s.pool.Close() +} func (s *PostgresSink) Name() string { return "postgres" } func (s *PostgresSink) Health(ctx context.Context) error { return s.pool.Ping(ctx) } func (s *PostgresSink) SaveJob(record *job.Record) { - select { - case s.queue <- record: - default: - } -} - -func (s *PostgresSink) run() { - for record := range s.queue { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - _ = s.upsert(ctx, record) - cancel() - } + s.writer.enqueue(record) } func migrate(ctx context.Context, pool *pgxpool.Pool) error { From 4669cea783927291eaef48ec11c38e8ef1c673bd Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 09:19:06 +0200 Subject: [PATCH 11/15] chore: update Garage to 2.4.1 --- docker-compose.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker-compose.yml b/docker-compose.yml index e492318..168dff4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -19,7 +19,7 @@ services: - "56379:6379" garage: - image: dxflrs/garage:v2.2.0 + image: dxflrs/garage:v2.4.1 ports: - "3900:3900" - "3901:3901" From 43c9520165779428f3b979446e20debfa43561d9 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 09:43:04 +0200 Subject: [PATCH 12/15] fix: wait for downloader workers before closing persistence --- cmd/server/lifecycle.go | 29 +++++++++ cmd/server/lifecycle_test.go | 71 ++++++++++++++++++++++ cmd/server/main.go | 41 +++++++------ internal/pipeline/lifecycle.go | 70 ++++++++++++++++++++++ internal/pipeline/lifecycle_test.go | 91 +++++++++++++++++++++++++++++ internal/pipeline/runner.go | 53 +---------------- 6 files changed, 283 insertions(+), 72 deletions(-) create mode 100644 cmd/server/lifecycle.go create mode 100644 cmd/server/lifecycle_test.go create mode 100644 internal/pipeline/lifecycle.go create mode 100644 internal/pipeline/lifecycle_test.go diff --git a/cmd/server/lifecycle.go b/cmd/server/lifecycle.go new file mode 100644 index 0000000..e89b1e4 --- /dev/null +++ b/cmd/server/lifecycle.go @@ -0,0 +1,29 @@ +package main + +import ( + "context" + "errors" + "net/http" + "time" +) + +func serve(ctx context.Context, server *http.Server) error { + result := make(chan error, 1) + go func() { result <- server.ListenAndServe() }() + select { + case err := <-result: + return err + case <-ctx.Done(): + } + shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + shutdownErr := server.Shutdown(shutdownCtx) + if shutdownErr != nil { + shutdownErr = errors.Join(shutdownErr, server.Close()) + } + listenErr := <-result + if errors.Is(listenErr, http.ErrServerClosed) { + listenErr = nil + } + return errors.Join(shutdownErr, listenErr) +} diff --git a/cmd/server/lifecycle_test.go b/cmd/server/lifecycle_test.go new file mode 100644 index 0000000..5e695d9 --- /dev/null +++ b/cmd/server/lifecycle_test.go @@ -0,0 +1,71 @@ +package main + +import ( + "context" + "net" + "net/http" + "testing" + "time" +) + +func TestServeWaitsForActiveHTTPHandler(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + server := &http.Server{Addr: "127.0.0.1:0", Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(entered) + <-release + w.WriteHeader(http.StatusNoContent) + })} + listening := make(chan net.Addr, 1) + server.BaseContext = func(listener net.Listener) context.Context { + listening <- listener.Addr() + return context.Background() + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + finished := make(chan error, 1) + go func() { finished <- serve(ctx, server) }() + var addr net.Addr + select { + case addr = <-listening: + case <-time.After(3 * time.Second): + t.Fatal("server did not listen") + } + response := make(chan error, 1) + go func() { + res, err := http.Get("http://" + addr.String()) + if err == nil { + res.Body.Close() + } + response <- err + }() + <-entered + cancel() + select { + case err := <-finished: + t.Fatalf("shutdown returned before handler: %v", err) + case <-time.After(20 * time.Millisecond): + } + close(release) + if err := <-response; err != nil { + t.Fatal(err) + } + select { + case err := <-finished: + if err != nil { + t.Fatal(err) + } + case <-time.After(3 * time.Second): + t.Fatal("shutdown did not finish") + } +} + +func TestServeReturnsListenFailure(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + if err := serve(t.Context(), &http.Server{Addr: listener.Addr().String()}); err == nil { + t.Fatal("expected address in use") + } +} diff --git a/cmd/server/main.go b/cmd/server/main.go index 2865276..c5505a5 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -2,7 +2,6 @@ package main import ( "context" - "errors" "log/slog" "net/http" "os" @@ -21,19 +20,22 @@ import ( ) func main() { - cfg := config.Load() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + if err := run(ctx, config.Load()); err != nil { + slog.Error("server failed", "error", err) + os.Exit(1) + } +} +func run(ctx context.Context, cfg config.Config) error { files, err := artifactStore(cfg) if err != nil { - slog.Error("artifact store failed", "error", err) - os.Exit(1) + return err } sinks, health, restored, pending, cleanupFn, err := sinks(ctx, cfg) if err != nil { - slog.Error("persistence failed", "error", err) - os.Exit(1) + return err } defer cleanupFn() store := job.NewStore(cfg.PublicBaseURL, sinks...) @@ -41,12 +43,16 @@ func main() { pendingIDs := store.RestorePending(pending) disk, err := storage.NewMonitor(cfg.DataDir, cfg.MinFreeBytes, cfg.MinFreePercent) if err != nil { - slog.Error("storage monitor failed", "error", err) - os.Exit(1) + return err } + workerCtx, cancelWorkers := context.WithCancel(ctx) runner := pipeline.NewRunner(cfg, store, files, disk) - runner.Start(ctx) - cleanup.Start(ctx, cfg.DataDir, cfg.StorageBackend) + runner.Start(workerCtx) + defer func() { + cancelWorkers() + runner.Wait() + }() + cleanup.Start(workerCtx, cfg.DataDir, cfg.StorageBackend) for _, id := range pendingIDs { if err := runner.EnqueueBlocking(ctx, id); err != nil { slog.Warn("failed to restore queued job", "id", id, "error", err) @@ -56,18 +62,8 @@ func main() { health = append(health, files) handler := api.NewServer(store, runner, files, disk, health...).Routes() server := &http.Server{Addr: cfg.HTTPAddr, Handler: handler, ReadHeaderTimeout: 10 * time.Second} - go func() { - <-ctx.Done() - shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() - _ = server.Shutdown(shutdownCtx) - }() - slog.Info("server listening", "addr", cfg.HTTPAddr) - if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("server failed", "error", err) - os.Exit(1) - } + return serve(ctx, server) } func artifactStore(cfg config.Config) (artifact.Store, error) { @@ -99,16 +95,19 @@ func sinks(ctx context.Context, cfg config.Config) ([]job.Sink, []api.HealthChec } postgres, err := db.OpenPostgres(ctx, cfg.DatabaseURL) if err != nil { + cleanup() return nil, nil, nil, nil, nil, err } restored, err := postgres.LoadDone(ctx) if err != nil { postgres.Close() + cleanup() return nil, nil, nil, nil, nil, err } pending, err := postgres.LoadRunnable(ctx) if err != nil { postgres.Close() + cleanup() return nil, nil, nil, nil, nil, err } out = append(out, postgres) diff --git a/internal/pipeline/lifecycle.go b/internal/pipeline/lifecycle.go new file mode 100644 index 0000000..e92657a --- /dev/null +++ b/internal/pipeline/lifecycle.go @@ -0,0 +1,70 @@ +package pipeline + +import ( + "context" + "fmt" + "log/slog" + + "typetype-downloader-go/internal/job" +) + +func (r *Runner) Start(ctx context.Context) { + for range r.cfg.MaxWorkers { + r.workers.Add(1) + go func() { + defer r.workers.Done() + r.worker(ctx) + }() + } +} + +func (r *Runner) Wait() { + r.workers.Wait() +} + +func (r *Runner) Enqueue(id string) error { + select { + case r.queue <- id: + return nil + default: + return fmt.Errorf("job queue is full") + } +} + +func (r *Runner) EnqueueBlocking(ctx context.Context, id string) error { + select { + case r.queue <- id: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func (r *Runner) worker(ctx context.Context) { + for ctx.Err() == nil { + select { + case <-ctx.Done(): + return + case id := <-r.queue: + if ctx.Err() != nil { + return + } + r.process(ctx, id) + } + } +} + +func (r *Runner) process(parent context.Context, id string) { + record, ok := r.store.Get(id) + if !ok || record.Status != job.StatusQueued { + return + } + ctx, cancel := context.WithCancel(parent) + r.store.Start(id, cancel) + defer cancel() + if err := r.run(ctx, id, record); err != nil { + code := failureCode(ctx, err) + r.store.Fail(id, code, err) + slog.Warn("job failed", "id", id, "error", err) + } +} diff --git a/internal/pipeline/lifecycle_test.go b/internal/pipeline/lifecycle_test.go new file mode 100644 index 0000000..9871549 --- /dev/null +++ b/internal/pipeline/lifecycle_test.go @@ -0,0 +1,91 @@ +package pipeline + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "typetype-downloader-go/internal/config" + "typetype-downloader-go/internal/job" +) + +type terminalSink struct { + entered chan struct{} + release chan struct{} + done chan struct{} +} + +func (s terminalSink) SaveJob(record *job.Record) { + if record.Status == job.StatusFailed { + close(s.entered) + <-s.release + close(s.done) + } +} + +func TestWaitIncludesFinalPersistenceNotification(t *testing.T) { + requested := make(chan struct{}) + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(requested) + <-r.Context().Done() + })) + defer upstream.Close() + sink := terminalSink{make(chan struct{}), make(chan struct{}), make(chan struct{})} + store := job.NewStore("", sink) + store.Restore([]*job.Record{{ID: "active", URL: "https://www.bilibili.com/video/BV1xx411c7mD", Status: job.StatusQueued}}) + runner := NewRunner(config.Config{MaxWorkers: 1, MaxQueueSize: 2, TypeTypeAPIBase: upstream.URL}, store, nil, nil) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + runner.Start(ctx) + if err := runner.Enqueue("active"); err != nil { + t.Fatal(err) + } + select { + case <-requested: + case <-time.After(3 * time.Second): + t.Fatal("worker did not start") + } + cancel() + select { + case <-sink.entered: + case <-time.After(3 * time.Second): + t.Fatal("worker did not persist cancellation") + } + waited := make(chan struct{}) + go func() { runner.Wait(); close(waited) }() + select { + case <-waited: + t.Fatal("Wait returned before final notification") + case <-time.After(20 * time.Millisecond): + } + close(sink.release) + select { + case <-waited: + case <-time.After(3 * time.Second): + t.Fatal("worker did not stop") + } + select { + case <-sink.done: + default: + t.Fatal("final notification missing") + } +} + +func TestCancelledRunnerLeavesQueuedJobsForRestart(t *testing.T) { + store := job.NewStore("") + store.Restore([]*job.Record{{ID: "queued", Status: job.StatusQueued}}) + runner := NewRunner(config.Config{MaxWorkers: 2, MaxQueueSize: 2}, store, nil, nil) + if err := runner.Enqueue("queued"); err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(t.Context()) + cancel() + runner.Start(ctx) + runner.Wait() + record, _ := store.Get("queued") + if record.Status != job.StatusQueued { + t.Fatalf("queued job changed: %s", record.Status) + } +} diff --git a/internal/pipeline/runner.go b/internal/pipeline/runner.go index 544559b..708d5b4 100644 --- a/internal/pipeline/runner.go +++ b/internal/pipeline/runner.go @@ -2,11 +2,11 @@ package pipeline import ( "context" - "fmt" "log/slog" "net/http" "os" "path/filepath" + "sync" "time" "typetype-downloader-go/internal/artifact" @@ -25,6 +25,7 @@ type Runner struct { disk *storage.Monitor http *http.Client queue chan string + workers sync.WaitGroup } func NewRunner(cfg config.Config, store *job.Store, files artifact.Store, disk *storage.Monitor) *Runner { @@ -39,56 +40,6 @@ func NewRunner(cfg config.Config, store *job.Store, files artifact.Store, disk * } } -func (r *Runner) Start(ctx context.Context) { - for range r.cfg.MaxWorkers { - go r.worker(ctx) - } -} - -func (r *Runner) Enqueue(id string) error { - select { - case r.queue <- id: - return nil - default: - return fmt.Errorf("job queue is full") - } -} - -func (r *Runner) EnqueueBlocking(ctx context.Context, id string) error { - select { - case r.queue <- id: - return nil - case <-ctx.Done(): - return ctx.Err() - } -} - -func (r *Runner) worker(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case id := <-r.queue: - r.process(ctx, id) - } - } -} - -func (r *Runner) process(parent context.Context, id string) { - record, ok := r.store.Get(id) - if !ok || record.Status != job.StatusQueued { - return - } - ctx, cancel := context.WithCancel(parent) - r.store.Start(id, cancel) - defer cancel() - if err := r.run(ctx, id, record); err != nil { - code := failureCode(ctx, err) - r.store.Fail(id, code, err) - slog.Warn("job failed", "id", id, "error", err) - } -} - func (r *Runner) run(ctx context.Context, id string, record *job.Record) error { started := time.Now() var stream *typetype.StreamResponse From fe23d4b60dcaafbc60add113950a05d8d80f7e90 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 09:48:42 +0200 Subject: [PATCH 13/15] fix: make downloader job snapshots race free --- internal/job/clone.go | 11 +++++-- internal/job/response.go | 12 +++---- internal/job/store.go | 23 +++++++------- internal/job/store_test.go | 65 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 92 insertions(+), 19 deletions(-) diff --git a/internal/job/clone.go b/internal/job/clone.go index f848597..acf7880 100644 --- a/internal/job/clone.go +++ b/internal/job/clone.go @@ -11,12 +11,19 @@ func cloneRecord(record *Record) *Record { copy.MuxMs = cloneInt64(record.MuxMs) copy.TotalMs = cloneInt64(record.TotalMs) if record.Resolved != nil { - resolved := *record.Resolved - copy.Resolved = &resolved + copy.Resolved = cloneResolved(record.Resolved) } return © } +func cloneResolved(value *ResolvedOutput) *ResolvedOutput { + if value == nil { + return nil + } + copy := *value + return © +} + func cloneString(value *string) *string { if value == nil { return nil diff --git a/internal/job/response.go b/internal/job/response.go index fafdae8..c9c6491 100644 --- a/internal/job/response.go +++ b/internal/job/response.go @@ -20,15 +20,15 @@ func (s *Store) toResponse(record *Record) Response { URL: record.URL, Status: record.Status, Title: record.Title, - Error: record.Error, - ErrorCode: record.ErrorCode, - Resolved: record.Resolved, + Error: cloneString(record.Error), + ErrorCode: cloneString(record.ErrorCode), + Resolved: cloneResolved(record.Resolved), QueuedAt: formatTime(record.QueuedAt), StartedAt: formatTimePtr(record.StartedAt), FinishedAt: formatTimePtr(record.FinishedAt), - DownloadMs: record.DownloadMs, - MuxMs: record.MuxMs, - TotalMs: record.TotalMs, + DownloadMs: cloneInt64(record.DownloadMs), + MuxMs: cloneInt64(record.MuxMs), + TotalMs: cloneInt64(record.TotalMs), } if record.Artifact != "" { artifactURL := s.baseURL + "/jobs/" + record.ID + "/artifact" diff --git a/internal/job/store.go b/internal/job/store.go index 68e7813..ca97f57 100644 --- a/internal/job/store.go +++ b/internal/job/store.go @@ -71,22 +71,20 @@ func (s *Store) Create(rawURL string, options Options, authorization string) (*R if err != nil { return nil, false, false, err } - s.mu.RLock() - if existingID := s.cache[cacheKey]; existingID != "" { - if existing := s.jobs[existingID]; existing != nil && existing.Status != StatusFailed { - cached := existing.Status == StatusDone - record := cloneRecord(existing) - s.mu.RUnlock() - return record, cached, false, nil - } - } - s.mu.RUnlock() id, err := newID() if err != nil { return nil, false, false, err } record := &Record{ID: id, CacheKey: cacheKey, URL: rawURL, Authorization: authorization, Options: options, Status: StatusQueued, QueuedAt: time.Now()} s.mu.Lock() + if existingID := s.cache[cacheKey]; existingID != "" { + if existing := s.jobs[existingID]; existing != nil && existing.Status != StatusFailed { + cached := existing.Status == StatusDone + snapshot := cloneRecord(existing) + s.mu.Unlock() + return snapshot, cached, false, nil + } + } s.jobs[id] = record s.cache[cacheKey] = id s.mu.Unlock() @@ -98,11 +96,14 @@ func (s *Store) Create(rawURL string, options Options, authorization string) (*R func (s *Store) Get(id string) (*Record, bool) { s.mu.RLock() record, ok := s.jobs[id] + if ok { + record = cloneRecord(record) + } s.mu.RUnlock() if !ok { return nil, false } - return cloneRecord(record), true + return record, true } func (s *Store) Start(id string, cancel func()) bool { diff --git a/internal/job/store_test.go b/internal/job/store_test.go index 935e1ef..8a2d43e 100644 --- a/internal/job/store_test.go +++ b/internal/job/store_test.go @@ -2,6 +2,7 @@ package job import ( "errors" + "sync" "testing" ) @@ -29,6 +30,70 @@ func TestStoreCreateStartProgressDone(t *testing.T) { } } +func TestStoreCreateIsAtomicForConcurrentDuplicateRequests(t *testing.T) { + store := NewStore("http://localhost") + const callers = 32 + start := make(chan struct{}) + results := make(chan *Record, callers) + created := make(chan bool, callers) + var group sync.WaitGroup + for range callers { + group.Add(1) + go func() { + defer group.Done() + <-start + record, _, wasCreated, err := store.Create("https://example.com/watch?v=duplicate", Options{Container: "mp4"}, "") + if err != nil { + t.Error(err) + return + } + results <- record + created <- wasCreated + }() + } + close(start) + group.Wait() + close(results) + close(created) + var firstID string + createdCount := 0 + for record := range results { + if firstID == "" { + firstID = record.ID + } + if record.ID != firstID { + t.Fatalf("duplicate request created %s and %s", firstID, record.ID) + } + } + for wasCreated := range created { + if wasCreated { + createdCount++ + } + } + if createdCount != 1 { + t.Fatalf("created count = %d, want 1", createdCount) + } +} + +func TestStoreGetAndResponseReturnSnapshots(t *testing.T) { + store := NewStore("http://localhost") + record, _, _, err := store.Create("https://example.com/watch?v=snapshot", Options{}, "") + if err != nil { + t.Fatal(err) + } + message := "original" + store.Fail(record.ID, "failure", errors.New(message)) + snapshot, ok := store.Get(record.ID) + if !ok || snapshot.Error == nil { + t.Fatal("missing error snapshot") + } + *snapshot.Error = "changed" + response, ok := store.Response(record.ID) + if !ok || response.Error == nil || *response.Error != message { + t.Fatalf("response error = %#v", response.Error) + } +} + func TestStoreCancelQueuedMarksFailed(t *testing.T) { store := NewStore("http://localhost") record, _, _, err := store.Create("https://example.com/watch?v=1", Options{}, "") From 6ca21e292ad9d88f8ba45e0312f0d2677aa030f7 Mon Sep 17 00:00:00 2001 From: Priveetee Date: Wed, 9 Sep 2026 10:43:39 +0200 Subject: [PATCH 14/15] chore: update downloader Go toolchain --- Dockerfile | 2 +- go.mod | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Dockerfile b/Dockerfile index 900eaa5..85ebe66 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.26-alpine3.23 AS build +FROM golang:1.27-alpine3.23 AS build ARG BUILD_VERSION=1.7.0-dev ARG BUILD_REVISION=development diff --git a/go.mod b/go.mod index 54c2c74..92e531e 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module typetype-downloader-go -go 1.26.0 +go 1.27.0 require ( github.com/jackc/pgx/v5 v5.11.0 From c517ae8a8243db4b48ed5b2a214864e65ad6b4db Mon Sep 17 00:00:00 2001 From: Priveetee Date: Thu, 10 Sep 2026 22:41:57 +0200 Subject: [PATCH 15/15] test: avoid flaky storage reservation boundary --- internal/storage/monitor_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/storage/monitor_test.go b/internal/storage/monitor_test.go index dbdc7d6..346dbe1 100644 --- a/internal/storage/monitor_test.go +++ b/internal/storage/monitor_test.go @@ -41,7 +41,7 @@ func TestMonitorUsesPercentageThreshold(t *testing.T) { } func TestMonitorTracksAndReleasesReservations(t *testing.T) { - monitor, err := NewMonitor(t.TempDir(), 1, 1) + monitor, err := NewMonitor(t.TempDir(), 1, 0) if err != nil { t.Fatal(err) } @@ -49,7 +49,7 @@ func TestMonitorTracksAndReleasesReservations(t *testing.T) { if err != nil { t.Fatal(err) } - reserved := capacity.FreeBytes - capacity.RequiredFreeBytes + const reserved = uint64(1 << 20) release, err := monitor.Reserve("job", reserved) if err != nil { t.Fatal(err)