Add github_alpine metadata-only package type (#115)

## Why

`github_alpine` is the Alpine/apk analog of the existing `github_deb`/`github_rpm` metadata-only remotes. It lets a plain GitHub-releases repo of `.apk` files be consumed as a real apk repository without artifactapi ever precaching the packages: it scans the repo's releases, derives each package's `.PKGINFO` from a ranged prefix fetch, synthesizes a per-arch `APKINDEX.tar.gz` from the cached metadata, and redirects the actual `.apk` downloads to a backend `releases_remote`. It stacks on the apk-local branch, reusing that work's alpine APKINDEX generator, `.apk`/`.PKGINFO` parser, Q1 pull-checksum, and `AlpineMetadata` store.

## How

- **`pkg/models`**: add `PackageGitHubAlpine` to the enum + validators (and test).
- **`internal/provider/alpine/github.go`**: the `github_alpine` provider. `ServeRemote` serves per-arch `<arch>/APKINDEX.tar.gz` (reusing `generateAPKIndex` over arch-filtered `AlpineMetadata` rows, `normalizeIndexPath` for apk's `./` dot-segment), 302-redirects `*.apk` to `{proxyBaseURL}/api/v1/remote/{releases_remote}/{path}`, and cold-starts with a 503 + `Retry-After`. `scanWithState` lists releases with ETag/If-None-Match and incrementally derives/prunes. `deriveAsset` does a **ranged GET of just the front of the `.apk`** — the control gzip stream carrying `.PKGINFO` sits near the front — doubling the range on truncation; it parses `.PKGINFO` and computes the `C:` Q1 checksum (`Q1`+base64(sha1(control stream))). `FilePath` = the github-relative asset path so the redirect resolves.
- **`internal/provider/alpine/syncer.go`**: a parallel background `Syncer` (own worker pool, shared rate limiter, deduped queue, DB lease), separate from the deb/rpm syncers.
- **`internal/database/alpine_github_sync.go`** + `github_alpine_sync_state` table: `ListGitHubAlpineRemotes` + Claim/Release per-remote sync lease, kept separate from the deb/rpm tables.
- **`internal/server/server.go`**: construct + `Run` the alpine syncer alongside deb/rpm and register it in the `PackageType→Primer` map (priming on create then flows through the existing generic `remotes.go` path).

The rpm/deb providers, syncers, and tables are untouched — this adds parallel alpine equivalents and reuses shared helpers already present in the alpine package.

## Tests

Mirror the deb github tests: scan derives `.PKGINFO`/Q1 from a ranged prefix of a `testsupport.MinimalApk` served over an httptest range server (no full download); diff/prune; pattern filter; per-arch `ServeRemote` routing (index served per-arch and grouped, dot-segment collapse, `.apk` → 302, cold-start 503, warm 200, canceled-request-serves-cache); DB lease prevents a second replica; prime bypasses the recency window. `go build`/`go vet`/`go mod tidy` clean, `make test` (-race) green, pre-commit green.

Reviewed-on: #115
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
This commit was merged in pull request #115.
This commit is contained in:
2026-08-12 20:48:42 +10:00
committed by BenVincent
parent 7f77666709
commit 5a06c16797
9 changed files with 1872 additions and 29 deletions
+71
View File
@@ -0,0 +1,71 @@
package database
import (
"context"
"errors"
"time"
"github.com/jackc/pgx/v5"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// ListGitHubAlpineRemotes returns every github_alpine remote so the syncer can
// sweep them on each poll tick.
func (db *DB) ListGitHubAlpineRemotes(ctx context.Context) ([]models.Remote, error) {
rows, err := db.Pool.Query(ctx, `SELECT `+remoteCols+` FROM remotes WHERE package_type = $1 ORDER BY name`, models.PackageGitHubAlpine)
if err != nil {
return nil, err
}
defer rows.Close()
var remotes []models.Remote
for rows.Next() {
var r models.Remote
if err := scanRemote(rows, &r); err != nil {
return nil, err
}
remotes = append(remotes, r)
}
return remotes, rows.Err()
}
// ClaimGitHubAlpineSyncLease atomically claims the per-remote sync lease. It
// succeeds only when the remote is due (never synced, or synced longer than
// freshness ago) and no live lease is held by another replica. A zero freshness
// (prime scans) ignores the recency gate. The returned etag is the stored
// releases-list ETag, shared across replicas.
func (db *DB) ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (bool, string, error) {
row := db.Pool.QueryRow(ctx, `
INSERT INTO github_alpine_sync_state AS s (remote_name, sync_lease_owner, sync_lease_expires)
VALUES ($1, $2, now() + make_interval(secs => $4))
ON CONFLICT (remote_name) DO UPDATE
SET sync_lease_owner = $2,
sync_lease_expires = now() + make_interval(secs => $4)
WHERE (s.last_synced_at IS NULL OR s.last_synced_at < now() - make_interval(secs => $3))
AND (s.sync_lease_expires IS NULL OR s.sync_lease_expires < now())
RETURNING s.etag
`, remoteName, owner, freshness.Seconds(), lease.Seconds())
var etag string
if err := row.Scan(&etag); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return false, "", nil
}
return false, "", err
}
return true, etag, nil
}
// ReleaseGitHubAlpineSyncLease records the completed scan and frees the lease.
// Only the owning replica may release; last_synced_at advances so the next poll
// waits a full freshness window, and etag is persisted for the next conditional
// request.
func (db *DB) ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error {
_, err := db.Pool.Exec(ctx, `
UPDATE github_alpine_sync_state
SET last_synced_at = $3, etag = $4, sync_lease_owner = '', sync_lease_expires = NULL
WHERE remote_name = $1 AND sync_lease_owner = $2
`, remoteName, owner, syncedAt, etag)
return err
}
+8
View File
@@ -227,6 +227,14 @@ func (db *DB) migrate() error {
sync_lease_expires TIMESTAMPTZ
);
CREATE TABLE IF NOT EXISTS github_alpine_sync_state (
remote_name TEXT PRIMARY KEY,
etag TEXT DEFAULT '',
last_synced_at TIMESTAMPTZ,
sync_lease_owner TEXT DEFAULT '',
sync_lease_expires TIMESTAMPTZ
);
CREATE TABLE IF NOT EXISTS signing_keys (
purpose TEXT PRIMARY KEY,
private_key_armor TEXT NOT NULL,
+714
View File
@@ -0,0 +1,714 @@
package alpine
import (
"bytes"
"compress/gzip"
"context"
"crypto/sha1"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"net/http"
"net/url"
"regexp"
"strings"
"sync"
"time"
"golang.org/x/time/rate"
"git.unkin.net/unkin/artifactapi/internal/githubauth"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// gitHubProvider is the process-wide singleton for github_alpine. The background
// Syncer binds its shared rate limiter and work queue onto this instance so the
// request path and the syncer drive the same derive machinery.
var gitHubProvider = newGitHubProvider()
func init() {
provider.Register(gitHubProvider)
}
// Tuning knobs for the no-precache control fetch. An .apk is up to three
// concatenated gzip streams (optional signature, control, data); the control
// stream carrying .PKGINFO sits near the front, so a small prefix reliably
// covers it.
const (
defaultHeaderRangeInitial = 32 << 10 // 32 KiB — covers the control stream of almost every .apk
defaultHeaderRangeMax = 16 << 20 // 16 MiB — give up past this and skip the asset
defaultReleasePageCap = 10 // 100 releases/page * 10 pages
defaultScanTimeout = 10 * time.Minute
defaultServeTimeout = 30 * time.Second
defaultColdWait = 8 * time.Second
)
// GitHubProvider is a metadata-only remote: it scans a GitHub repo's releases
// for .apk assets, derives per-asset .PKGINFO metadata via a ranged prefix fetch
// (never downloading whole packages), synthesizes a per-arch APKINDEX from that
// cached metadata, and redirects package downloads to a backend "releases_remote"
// (the generic github.com remote) that serves the actual bytes.
type GitHubProvider struct {
client *http.Client
headerInitial int64
headerMax int64
pageCap int
scanTimeout time.Duration
serveTimeout time.Duration
coldWait time.Duration
limiter *rate.Limiter
syncer *Syncer
serverCred githubauth.Credential
mu sync.Mutex
scanning map[string]bool
lastScan map[string]time.Time
}
func newGitHubProvider() *GitHubProvider {
return &GitHubProvider{
client: &http.Client{},
headerInitial: defaultHeaderRangeInitial,
headerMax: defaultHeaderRangeMax,
pageCap: defaultReleasePageCap,
scanTimeout: defaultScanTimeout,
serveTimeout: defaultServeTimeout,
coldWait: defaultColdWait,
scanning: map[string]bool{},
lastScan: map[string]time.Time{},
}
}
func (p *GitHubProvider) limiterWait(ctx context.Context) error {
if p.limiter == nil {
return nil
}
return p.limiter.Wait(ctx)
}
func (p *GitHubProvider) Type() models.PackageType { return models.PackageGitHubAlpine }
func (p *GitHubProvider) Classify(path string) provider.Mutability {
if strings.HasSuffix(path, "APKINDEX.tar.gz") {
return provider.Mutable
}
return provider.Immutable
}
func (p *GitHubProvider) ContentType(path string) string {
switch {
case strings.HasSuffix(path, ".apk"):
return "application/vnd.android.package-archive"
case strings.HasSuffix(path, ".tar.gz"):
return "application/gzip"
}
return "application/octet-stream"
}
func (p *GitHubProvider) UpstreamURL(remote models.Remote, path string) string {
return strings.TrimRight(remote.BaseURL, "/") + "/" + strings.TrimLeft(path, "/")
}
func (p *GitHubProvider) RewriteResponse(_ []byte, _ models.Remote, _ string) ([]byte, error) {
return nil, nil
}
func (p *GitHubProvider) AuthHeaders(ctx context.Context, remote models.Remote) (http.Header, error) {
return p.githubHeaders(ctx, remote, false)
}
// ServeRemote answers a request against a github_alpine remote. It refreshes the
// derived metadata (bounded by mutable_ttl), serves a synthesized per-arch
// APKINDEX.tar.gz, and 302-redirects .apk downloads to the backend
// releases_remote. Returns false only for paths it does not own.
func (p *GitHubProvider) ServeRemote(w http.ResponseWriter, r *http.Request, remote models.Remote, reqPath, proxyBaseURL string, store provider.RemoteMetadataStore) bool {
p.onRequest(remote, store)
// apk requests the index at "./<arch>/APKINDEX.tar.gz"; collapse the
// dot-segment before matching, mirroring the local indexer.
path := normalizeIndexPath(reqPath)
if strings.HasSuffix(path, "APKINDEX.tar.gz") {
p.serveIndex(w, r, remote, path, store)
return true
}
if strings.HasSuffix(path, ".apk") {
if remote.ReleasesRemote == "" {
http.Error(w, "github_alpine remote has no releases_remote configured for downloads", http.StatusInternalServerError)
return true
}
p.serveApkRedirect(w, r, remote, path, proxyBaseURL, store)
return true
}
return false
}
// serveApkRedirect resolves an apk-reconstructed download path — apk builds
// "<arch>/<name>-<version>.apk" itself because APKINDEX carries no filename — to
// the real github-relative asset path stored on the metadata row, then redirects
// to the backend releases_remote. Passing the inbound path through verbatim would
// point at a nonexistent, allowlist-denied github.com path.
func (p *GitHubProvider) serveApkRedirect(w http.ResponseWriter, r *http.Request, remote models.Remote, path, proxyBaseURL string, store provider.RemoteMetadataStore) {
arch := strings.TrimSuffix(path[:strings.LastIndex(path, "/")+1], "/")
basename := path[strings.LastIndex(path, "/")+1:]
if arch == "" || strings.Contains(arch, "/") {
http.Error(w, "apk download must be requested per-arch: <arch>/<name>-<version>.apk", http.StatusNotFound)
return
}
reader, ok := store.(provider.AlpineMetadataReader)
if !ok {
http.Error(w, "alpine metadata not available", http.StatusInternalServerError)
return
}
sctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), p.serveTimeout)
defer cancel()
rows, err := reader.ListAlpineMetadataEntries(sctx, remote.Name)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
for _, row := range rows {
if row.Arch == arch && row.Name+"-"+row.Version+".apk" == basename {
loc := strings.TrimRight(proxyBaseURL, "/") + "/api/v1/remote/" + remote.ReleasesRemote + "/" + strings.TrimLeft(row.FilePath, "/")
http.Redirect(w, r, loc, http.StatusFound)
return
}
}
http.Error(w, "package not found", http.StatusNotFound)
}
func (p *GitHubProvider) serveIndex(w http.ResponseWriter, r *http.Request, remote models.Remote, path string, store provider.RemoteMetadataStore) {
arch := strings.TrimSuffix(path, "APKINDEX.tar.gz")
arch = strings.Trim(arch, "/")
if arch == "" || strings.Contains(arch, "/") {
http.Error(w, "APKINDEX must be requested per-arch: <arch>/APKINDEX.tar.gz", http.StatusNotFound)
return
}
// Serve on a context detached from the inbound request so a client disconnect
// never cancels the metadata DB read and surfaces as a 500.
sctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), p.serveTimeout)
defer cancel()
if p.syncer != nil && !p.ensurePrimed(sctx, remote, store) {
w.Header().Set("Retry-After", "5")
http.Error(w, "metadata is being prepared, retry shortly", http.StatusServiceUnavailable)
return
}
reader, ok := store.(provider.AlpineMetadataReader)
if !ok {
http.Error(w, "alpine metadata not available", http.StatusInternalServerError)
return
}
metas, err := reader.ListAlpineMetadataEntries(sctx, remote.Name)
if err != nil {
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
http.Error(w, "metadata read canceled", http.StatusServiceUnavailable)
return
}
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
var filtered []provider.AlpineMetadata
for _, m := range metas {
if m.Arch == arch {
filtered = append(filtered, m)
}
}
w.Header().Set("Content-Type", "application/gzip")
w.WriteHeader(http.StatusOK)
w.Write(generateAPKIndex(filtered))
}
// onRequest keeps a remote's derived metadata fresh off the request path.
func (p *GitHubProvider) onRequest(remote models.Remote, store provider.RemoteMetadataStore) {
if p.syncer != nil {
p.syncer.enqueue(remote, false)
return
}
p.refresh(remote, store)
}
// ensurePrimed returns true once the remote has at least one cached row. On an
// empty cache it enqueues a prime and polls briefly for it to land.
func (p *GitHubProvider) ensurePrimed(ctx context.Context, remote models.Remote, store provider.RemoteMetadataStore) bool {
if !p.cacheEmpty(ctx, store, remote.Name) {
return true
}
if p.syncer != nil {
p.syncer.enqueue(remote, true)
}
deadline := time.Now().Add(p.coldWait)
for time.Now().Before(deadline) {
select {
case <-ctx.Done():
return false
case <-time.After(400 * time.Millisecond):
}
if !p.cacheEmpty(ctx, store, remote.Name) {
return true
}
}
return false
}
func (p *GitHubProvider) cacheEmpty(ctx context.Context, store provider.RemoteMetadataStore, name string) bool {
reader, ok := store.(provider.AlpineMetadataReader)
if !ok {
return false
}
rows, err := reader.ListAlpineMetadataEntries(ctx, name)
if err != nil {
return false
}
return len(rows) == 0
}
// refresh brings the derived metadata up to date without coupling the scan to
// the inbound request (legacy inline path used without a syncer / in unit tests).
func (p *GitHubProvider) refresh(remote models.Remote, store provider.RemoteMetadataStore) {
ttl := time.Duration(remote.MutableTTL) * time.Second
if ttl <= 0 {
ttl = 5 * time.Minute
}
p.mu.Lock()
last, ok := p.lastScan[remote.Name]
fresh := ok && time.Since(last) < ttl
if fresh || p.scanning[remote.Name] {
p.mu.Unlock()
return
}
p.scanning[remote.Name] = true
p.mu.Unlock()
if p.cacheEmpty(context.Background(), store, remote.Name) {
p.runScan(remote, store)
return
}
go p.runScan(remote, store)
}
func (p *GitHubProvider) runScan(remote models.Remote, store provider.RemoteMetadataStore) {
defer func() {
p.mu.Lock()
delete(p.scanning, remote.Name)
p.mu.Unlock()
}()
ctx, cancel := context.WithTimeout(context.Background(), p.scanTimeout)
defer cancel()
if err := p.scan(ctx, remote, store); err != nil {
slog.Error("github_alpine: release scan failed", "remote", remote.Name, "error", err)
return
}
p.mu.Lock()
p.lastScan[remote.Name] = time.Now()
p.mu.Unlock()
}
// scan runs a full unconditional derive. Retained for the legacy inline refresh
// path and existing tests; the syncer uses scanWithState.
func (p *GitHubProvider) scan(ctx context.Context, remote models.Remote, store provider.RemoteMetadataStore) error {
_, _, err := p.scanWithState(ctx, remote, store, "")
return err
}
// scanWithState derives metadata incrementally. It sends the prior releases-list
// ETag as a conditional request: a 304 means nothing changed. On a 200 it diffs
// the release assets against the cache, derives only new/changed assets, prunes
// assets that disappeared, and returns the new ETag.
func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote, store provider.RemoteMetadataStore, etag string) (newEtag string, changed bool, err error) {
inserter, ok := store.(provider.AlpineMetadataStore)
if !ok {
return etag, false, errors.New("store does not support alpine metadata writes")
}
deleter, ok := store.(provider.AlpineMetadataDeleter)
if !ok {
return etag, false, errors.New("store does not support alpine metadata deletes")
}
reader, ok := store.(provider.AlpineMetadataReader)
if !ok {
return etag, false, errors.New("store does not support alpine metadata reads")
}
releases, newEtag, notModified, err := p.fetchReleases(ctx, remote, etag)
if err != nil {
return etag, false, err
}
if notModified {
return etag, false, nil
}
existing, err := reader.ListAlpineMetadataEntries(ctx, remote.Name)
if err != nil {
return newEtag, false, err
}
existingByPath := make(map[string]provider.AlpineMetadata, len(existing))
for _, m := range existing {
existingByPath[m.FilePath] = m
}
allow, err := compilePatterns(remote.Patterns)
if err != nil {
return newEtag, false, err
}
seen := map[string]bool{}
for _, rel := range releases {
if rel.Draft {
continue
}
for _, asset := range rel.Assets {
if !strings.HasSuffix(strings.ToLower(asset.Name), ".apk") {
continue
}
if !matchesAny(allow, asset.Name) {
continue
}
fp := assetPath(asset)
if fp == "" {
continue
}
seen[fp] = true
if cur, ok := existingByPath[fp]; ok {
if asset.Digest == "" || cur.ContentHash == asset.Digest {
continue
}
_ = deleter.DeleteAlpineMetadata(ctx, remote.Name, fp)
}
meta, err := p.deriveAsset(ctx, remote, asset, fp)
if err != nil {
slog.Warn("github_alpine: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err)
continue
}
if err := inserter.InsertAlpineMetadata(ctx, meta); err != nil {
slog.Error("github_alpine: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err)
continue
}
slog.Info("github_alpine: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Arch)
}
}
for fp := range existingByPath {
if !seen[fp] {
_ = deleter.DeleteAlpineMetadata(ctx, remote.Name, fp)
}
}
return newEtag, true, nil
}
type ghRelease struct {
TagName string `json:"tag_name"`
Draft bool `json:"draft"`
Assets []ghAsset `json:"assets"`
}
type ghAsset struct {
Name string `json:"name"`
Size int64 `json:"size"`
BrowserDownloadURL string `json:"browser_download_url"`
Digest string `json:"digest"`
}
// fetchReleases lists a repo's releases, sending the prior ETag as If-None-Match
// on page 1 so an unchanged repo short-circuits to notModified. Every call waits
// on the shared limiter first.
func (p *GitHubProvider) fetchReleases(ctx context.Context, remote models.Remote, etag string) (all []ghRelease, newEtag string, notModified bool, err error) {
base := strings.TrimRight(remote.BaseURL, "/") + "/releases"
for page := 1; page <= p.pageCap; page++ {
u := fmt.Sprintf("%s?per_page=100&page=%d", base, page)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
if err != nil {
return nil, "", false, err
}
hdr, err := p.githubHeaders(ctx, remote, true)
if err != nil {
return nil, "", false, err
}
copyHeaders(req, hdr)
if page == 1 && etag != "" {
req.Header.Set("If-None-Match", etag)
}
if err := p.limiterWait(ctx); err != nil {
return nil, "", false, err
}
resp, err := p.client.Do(req)
if err != nil {
return nil, "", false, err
}
if page == 1 && resp.StatusCode == http.StatusNotModified {
io.Copy(io.Discard, resp.Body)
resp.Body.Close()
return nil, etag, true, nil
}
body, err := io.ReadAll(resp.Body)
respEtag := resp.Header.Get("ETag")
resp.Body.Close()
if err != nil {
return nil, "", false, err
}
if resp.StatusCode != http.StatusOK {
return nil, "", false, fmt.Errorf("github releases API %s: status %d", u, resp.StatusCode)
}
if page == 1 {
newEtag = respEtag
}
var releases []ghRelease
if err := json.Unmarshal(body, &releases); err != nil {
return nil, "", false, fmt.Errorf("decode releases: %w", err)
}
if len(releases) == 0 {
break
}
all = append(all, releases...)
if len(releases) < 100 {
break
}
}
return all, newEtag, false, nil
}
func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote, asset ghAsset, fp string) (*provider.AlpineMetadata, error) {
meta, err := p.fetchPkginfo(ctx, remote, asset.BrowserDownloadURL)
if err != nil {
return nil, err
}
if meta.Name == "" || meta.Arch == "" {
return nil, errors.New(".PKGINFO missing pkgname/arch")
}
meta.RepoName = remote.Name
meta.FilePath = fp
// S: the on-disk .apk size comes straight from the releases API, so we never
// download the body just to size it.
meta.DownloadSize = asset.Size
// ContentHash records the GitHub asset digest (when present) purely so the
// next scan can detect a changed asset; unlike deb it is not the index
// checksum (that is the Q1 control-stream sum already set in fetchPkginfo).
if asset.Digest != "" {
meta.ContentHash = asset.Digest
}
return meta, nil
}
// fetchPkginfo pulls only the front of the .apk with a ranged GET and derives the
// .PKGINFO fields plus the apk pull checksum (C: = Q1 + base64(sha1(control gzip
// stream))). The control stream sits near the front, so a small prefix suffices;
// a prefix that truncates it doubles the range and retries.
func (p *GitHubProvider) fetchPkginfo(ctx context.Context, remote models.Remote, downloadURL string) (*provider.AlpineMetadata, error) {
n := p.headerInitial
for {
body, full, err := p.rangeGet(ctx, remote, downloadURL, n)
if err != nil {
return nil, err
}
meta, complete, perr := pkginfoFromPrefix(body)
if perr != nil {
return nil, fmt.Errorf("parse apk .PKGINFO: %w", perr)
}
if complete {
return meta, nil
}
if full || n >= p.headerMax {
return nil, fmt.Errorf(".PKGINFO not found within %d bytes of %s", n, downloadURL)
}
n *= 2
if n > p.headerMax {
n = p.headerMax
}
}
}
// pkginfoFromPrefix parses the concatenated gzip streams present in a front
// prefix of an .apk. It walks each fully-covered gzip member until it finds the
// control stream (the one whose tar carries .PKGINFO), computes the Q1 pull
// checksum from that stream's raw bytes, and reads the .PKGINFO fields. A prefix
// too short to fully cover the control stream returns complete=false so the
// caller can widen the range.
func pkginfoFromPrefix(prefix []byte) (meta *provider.AlpineMetadata, complete bool, err error) {
br := bytes.NewReader(prefix)
zr, zerr := gzip.NewReader(br)
if zerr != nil {
if zerr == io.EOF || zerr == io.ErrUnexpectedEOF {
return nil, false, nil
}
return nil, false, zerr
}
prev := 0
for {
zr.Multistream(false)
out, rerr := io.ReadAll(zr)
if rerr != nil {
// A member truncated by the range boundary is not an error — widen.
if rerr == io.ErrUnexpectedEOF || rerr == io.EOF {
return nil, false, nil
}
return nil, false, rerr
}
end := len(prefix) - br.Len()
raw := prefix[prev:end]
if pkginfo, ok := pkginfoFromTar(out); ok {
m := parsePkginfo(pkginfo)
sum := sha1.Sum(raw)
m.Checksum = "Q1" + base64.StdEncoding.EncodeToString(sum[:])
return m, true, nil
}
prev = end
if rsterr := zr.Reset(br); rsterr != nil {
if rsterr == io.EOF {
// No more complete members in the prefix; the control stream is
// either not covered yet or genuinely absent — let the caller
// decide by widening (or hitting the full-object guard).
return nil, false, nil
}
if rsterr == io.ErrUnexpectedEOF {
return nil, false, nil
}
return nil, false, rsterr
}
}
}
// rangeGet returns the first n bytes of downloadURL. full is true when the
// response body was shorter than n (i.e. we already have the whole object).
func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
if err != nil {
return nil, false, err
}
hdr, err := p.githubHeaders(ctx, remote, false)
if err != nil {
return nil, false, err
}
copyHeaders(req, hdr)
req.Header.Set("Range", fmt.Sprintf("bytes=0-%d", n-1))
if err := p.limiterWait(ctx); err != nil {
return nil, false, err
}
resp, err := p.client.Do(req)
if err != nil {
return nil, false, err
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
return nil, false, fmt.Errorf("range GET %s: status %d", downloadURL, resp.StatusCode)
}
body, err := io.ReadAll(io.LimitReader(resp.Body, n))
if err != nil {
return nil, false, err
}
full := int64(len(body)) < n
return body, full, nil
}
// assetPath is the package's location relative to github.com — the path the
// backend releases_remote (base https://github.com) proxies. It doubles as the
// alpine_metadata key and the redirect target, so an .apk download resolves back
// to this remote and redirects to the backend.
func assetPath(asset ghAsset) string {
u, err := url.Parse(asset.BrowserDownloadURL)
if err != nil {
return ""
}
return strings.TrimPrefix(u.Path, "/")
}
// githubHeaders builds the outbound headers for a GitHub request, attaching a
// bearer credential when one is available. A per-remote credential wins; absent
// that, the process-wide server credential is used; absent both, the request is
// unauthenticated.
func (p *GitHubProvider) githubHeaders(ctx context.Context, remote models.Remote, api bool) (http.Header, error) {
h := http.Header{}
if api {
h.Set("Accept", "application/vnd.github+json")
h.Set("X-GitHub-Api-Version", "2022-11-28")
}
tok, err := p.githubToken(ctx, remote)
if err != nil {
return nil, err
}
if tok != "" {
h.Set("Authorization", "Bearer "+tok)
}
return h, nil
}
// githubToken resolves the bearer token for a remote. Precedence: a per-remote
// credential (password, then username) overrides the server credential.
func (p *GitHubProvider) githubToken(ctx context.Context, remote models.Remote) (string, error) {
if remote.Password != "" {
return remote.Password, nil
}
if remote.Username != "" {
return remote.Username, nil
}
if c := p.serverCredential(); c != nil {
return c.Token(ctx)
}
return "", nil
}
func (p *GitHubProvider) serverCredential() githubauth.Credential {
if p.serverCred != nil {
return p.serverCred
}
return githubauth.Server()
}
func copyHeaders(req *http.Request, h http.Header) {
for k, vals := range h {
for _, v := range vals {
req.Header.Add(k, v)
}
}
}
func compilePatterns(patterns []string) ([]*regexp.Regexp, error) {
var out []*regexp.Regexp
for _, p := range patterns {
re, err := regexp.Compile(p)
if err != nil {
return nil, fmt.Errorf("invalid pattern %q: %w", p, err)
}
out = append(out, re)
}
return out, nil
}
func matchesAny(res []*regexp.Regexp, s string) bool {
if len(res) == 0 {
return true
}
for _, re := range res {
if re.MatchString(s) {
return true
}
}
return false
}
+497
View File
@@ -0,0 +1,497 @@
package alpine
import (
"archive/tar"
"bytes"
"compress/gzip"
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"sync"
"testing"
"time"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/internal/testsupport"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// fakeStore is an in-memory provider.RemoteMetadataStore + AlpineMetadata
// store/reader/deleter keyed by file_path, mirroring the (repo_name, file_path)
// uniqueness of the real alpine_metadata table.
type fakeStore struct {
mu sync.Mutex
rows map[string]provider.AlpineMetadata
}
func newFakeStore() *fakeStore { return &fakeStore{rows: map[string]provider.AlpineMetadata{}} }
func (f *fakeStore) InsertAlpineMetadata(_ context.Context, m *provider.AlpineMetadata) error {
f.mu.Lock()
defer f.mu.Unlock()
if _, ok := f.rows[m.FilePath]; ok {
return nil // ON CONFLICT DO NOTHING
}
f.rows[m.FilePath] = *m
return nil
}
func (f *fakeStore) DeleteAlpineMetadata(_ context.Context, _, filePath string) error {
f.mu.Lock()
defer f.mu.Unlock()
delete(f.rows, filePath)
return nil
}
func (f *fakeStore) ListAlpineMetadataEntries(ctx context.Context, _ string) ([]provider.AlpineMetadata, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
f.mu.Lock()
defer f.mu.Unlock()
out := make([]provider.AlpineMetadata, 0, len(f.rows))
for _, m := range f.rows {
out = append(out, m)
}
return out, nil
}
// The generic RemoteMetadataStore surface (rpm/deb) is unused by the alpine
// github provider but required to satisfy the interface passed to ServeRemote.
func (f *fakeStore) InsertRPMMetadata(context.Context, *provider.RPMMetadata) error { return nil }
func (f *fakeStore) DeleteRPMMetadata(context.Context, string, string) error { return nil }
func (f *fakeStore) ListRPMMetadataEntries(context.Context, string) ([]provider.RPMMetadata, error) {
return nil, nil
}
func (f *fakeStore) InsertDebMetadata(context.Context, *provider.DebMetadata) error { return nil }
func (f *fakeStore) DeleteDebMetadata(context.Context, string, string) error { return nil }
var _ provider.RemoteMetadataStore = (*fakeStore)(nil)
// githubFixture serves the releases API and the .apk asset downloads (with Range
// support) for a set of packages. digest controls whether the asset carries a
// sha256 digest (change-detection path) or not.
type githubFixture struct {
srv *httptest.Server
apkBytes map[string][]byte
rangeHit map[string]int
fullHit map[string]int
etag string
releasesHit int
notModHit int
releaseAuth string
assetAuth string
mu sync.Mutex
}
func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
t.Helper()
f := &githubFixture{
apkBytes: map[string][]byte{},
rangeHit: map[string]int{},
fullHit: map[string]int{},
}
f.apkBytes["demo-1.2.3-r0.apk"] = testsupport.MinimalApk("demo", "1.2.3-r0", "x86_64")
mux := http.NewServeMux()
mux.HandleFunc("/repos/acme/tools/releases", func(w http.ResponseWriter, r *http.Request) {
page := r.URL.Query().Get("page")
if page != "" && page != "1" {
w.Write([]byte("[]"))
return
}
f.mu.Lock()
f.releasesHit++
f.releaseAuth = r.Header.Get("Authorization")
etag := f.etag
if etag != "" && r.Header.Get("If-None-Match") == etag {
f.notModHit++
f.mu.Unlock()
w.WriteHeader(http.StatusNotModified)
return
}
f.mu.Unlock()
if etag != "" {
w.Header().Set("ETag", etag)
}
var assets []map[string]any
for name := range f.apkBytes {
a := map[string]any{
"name": name,
"size": len(f.apkBytes[name]),
"browser_download_url": f.srv.URL + "/acme/tools/releases/download/v1.2.3/" + name,
}
if withDigest {
sum := sha256.Sum256(f.apkBytes[name])
a["digest"] = "sha256:" + hex.EncodeToString(sum[:])
}
assets = append(assets, a)
}
rel := []map[string]any{{"tag_name": "v1.2.3", "draft": false, "assets": assets}}
json.NewEncoder(w).Encode(rel)
})
mux.HandleFunc("/acme/tools/releases/download/", func(w http.ResponseWriter, r *http.Request) {
name := r.URL.Path[strings.LastIndex(r.URL.Path, "/")+1:]
body, ok := f.apkBytes[name]
if !ok {
http.Error(w, "not found", 404)
return
}
rng := r.Header.Get("Range")
f.mu.Lock()
f.assetAuth = r.Header.Get("Authorization")
if rng != "" {
f.rangeHit[name]++
} else {
f.fullHit[name]++
}
f.mu.Unlock()
if rng == "" {
w.WriteHeader(200)
w.Write(body)
return
}
var end int
fmt.Sscanf(rng, "bytes=0-%d", &end)
if end >= len(body)-1 {
end = len(body) - 1
}
w.Header().Set("Content-Range", fmt.Sprintf("bytes 0-%d/%d", end, len(body)))
w.Header().Set("Content-Length", strconv.Itoa(end+1))
w.WriteHeader(http.StatusPartialContent)
w.Write(body[:end+1])
})
f.srv = httptest.NewServer(mux)
t.Cleanup(f.srv.Close)
return f
}
func (f *githubFixture) remote() models.Remote {
return models.Remote{
Name: "acme-apk",
PackageType: models.PackageGitHubAlpine,
BaseURL: f.srv.URL + "/repos/acme/tools",
ReleasesRemote: "github",
MutableTTL: 3600,
}
}
func newTestProvider() *GitHubProvider {
p := newGitHubProvider()
p.headerInitial = 32 // force the ranged-fetch retry loop against the tiny fixture
p.headerMax = 1 << 20
return p
}
const demoPath = "acme/tools/releases/download/v1.2.3/demo-1.2.3-r0.apk"
func TestGitHubScanDerivesPkginfoFromPrefix(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
if err := p.scan(context.Background(), fx.remote(), store); err != nil {
t.Fatalf("scan: %v", err)
}
metas, _ := store.ListAlpineMetadataEntries(context.Background(), "acme-apk")
if len(metas) != 1 {
t.Fatalf("want 1 metadata row, got %d", len(metas))
}
m := metas[0]
if m.Name != "demo" || m.Version != "1.2.3-r0" || m.Arch != "x86_64" {
t.Fatalf("bad .PKGINFO fields: %+v", m)
}
if m.FilePath != demoPath {
t.Fatalf("FilePath = %q, want %q", m.FilePath, demoPath)
}
if int(m.DownloadSize) != len(fx.apkBytes["demo-1.2.3-r0.apk"]) {
t.Fatalf("DownloadSize = %d, want %d", m.DownloadSize, len(fx.apkBytes["demo-1.2.3-r0.apk"]))
}
if !strings.HasPrefix(m.Checksum, "Q1") {
t.Fatalf("Checksum not a Q1 pull checksum: %q", m.Checksum)
}
// The C: checksum must equal Q1 over the raw control gzip stream, matching the
// local-upload parser applied to the same bytes.
want, err := parseApk(fx.apkBytes["demo-1.2.3-r0.apk"])
if err != nil {
t.Fatalf("reference parseApk: %v", err)
}
if m.Checksum != want.Checksum {
t.Fatalf("Checksum = %q, want %q (Q1 of control stream)", m.Checksum, want.Checksum)
}
if fx.fullHit["demo-1.2.3-r0.apk"] != 0 {
t.Fatalf("expected no full download, got %d", fx.fullHit["demo-1.2.3-r0.apk"])
}
if fx.rangeHit["demo-1.2.3-r0.apk"] == 0 {
t.Fatalf("expected ranged .PKGINFO fetch")
}
}
func TestGitHubServeRemoteIndexAndRedirect(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
const proxyBase = "https://artifactapi.example"
// The per-arch index is served and triggers the initial scan.
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-apk/x86_64/APKINDEX.tar.gz", nil)
if !p.ServeRemote(rec, req, remote, "x86_64/APKINDEX.tar.gz", proxyBase, store) {
t.Fatal("ServeRemote did not handle APKINDEX")
}
if rec.Code != 200 {
t.Fatalf("APKINDEX bad: code=%d body=%s", rec.Code, rec.Body.String())
}
idx := readAPKIndex(t, rec.Body.Bytes())
if !strings.Contains(idx, "P:demo") || !strings.Contains(idx, "A:x86_64") {
t.Fatalf("APKINDEX missing package record: %s", idx)
}
if !strings.Contains(idx, "C:Q1") {
t.Fatalf("APKINDEX missing pull checksum: %s", idx)
}
// A different arch yields an empty (but valid) index.
rec = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, "/x", nil)
if !p.ServeRemote(rec, req, remote, "aarch64/APKINDEX.tar.gz", proxyBase, store) {
t.Fatal("ServeRemote did not handle aarch64 APKINDEX")
}
if rec.Code != 200 {
t.Fatalf("empty-arch index bad: %d", rec.Code)
}
if got := readAPKIndex(t, rec.Body.Bytes()); strings.Contains(got, "P:demo") {
t.Fatalf("aarch64 index should not carry the x86_64 package: %s", got)
}
// An .apk request arrives in apk's reconstructed shape
// "<arch>/<name>-<version>.apk" (APKINDEX carries no filename), NOT as the
// github-relative FilePath. ServeRemote must resolve it back to the stored
// FilePath before redirecting to the backend releases_remote.
rec = httptest.NewRecorder()
req = httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-apk/x86_64/demo-1.2.3-r0.apk", nil)
if !p.ServeRemote(rec, req, remote, "x86_64/demo-1.2.3-r0.apk", proxyBase, store) {
t.Fatal("ServeRemote did not handle .apk")
}
if rec.Code != http.StatusFound {
t.Fatalf("want 302, got %d", rec.Code)
}
wantLoc := proxyBase + "/api/v1/remote/github/" + demoPath
if got := rec.Header().Get("Location"); got != wantLoc {
t.Fatalf("Location = %q, want %q (must be the stored FilePath, not the inbound path)", got, wantLoc)
}
}
// An apk download whose reconstructed "<arch>/<name>-<version>.apk" matches no
// cached row must 404, never redirect to a bad path.
func TestGitHubServeRemoteApkRedirectNotFound(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
// Warm the cache so the store is populated but lacks the requested package.
if err := p.scan(context.Background(), remote, store); err != nil {
t.Fatalf("warm scan: %v", err)
}
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-apk/x86_64/nope-9.9.9.apk", nil)
if !p.ServeRemote(rec, req, remote, "x86_64/nope-9.9.9.apk", "https://x", store) {
t.Fatal("ServeRemote did not handle .apk")
}
if rec.Code != http.StatusNotFound {
t.Fatalf("want 404 for unknown package, got %d (Location=%q)", rec.Code, rec.Header().Get("Location"))
}
}
// apk requests the index at "./<arch>/APKINDEX.tar.gz"; ServeRemote must collapse
// the dot-segment and synthesize the same index as the un-prefixed request.
func TestGitHubServeRemoteApkDotSegment(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
const proxyBase = "https://artifactapi.example"
serve := func(path string) *httptest.ResponseRecorder {
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-apk/"+path, nil)
if !p.ServeRemote(rec, req, remote, path, proxyBase, store) {
t.Fatalf("ServeRemote did not handle %q", path)
}
return rec
}
plain, dotted := serve("x86_64/APKINDEX.tar.gz"), serve("./x86_64/APKINDEX.tar.gz")
if plain.Code != 200 || dotted.Code != 200 {
t.Fatalf("index: plain=%d dotted=%d, want 200/200", plain.Code, dotted.Code)
}
if !bytes.Equal(plain.Body.Bytes(), dotted.Body.Bytes()) {
t.Error("./<arch>/APKINDEX.tar.gz body differs from the un-prefixed body")
}
}
func TestGitHubServeRemoteRejectsNonPerArchIndex(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/x", nil)
if !p.ServeRemote(rec, req, fx.remote(), "APKINDEX.tar.gz", "https://x", store) {
t.Fatal("expected handled")
}
if rec.Code != http.StatusNotFound {
t.Fatalf("bare APKINDEX must 404 (per-arch required), got %d", rec.Code)
}
}
// A canceled inbound request must still serve the warm cache (detached context),
// not turn the metadata read into a 500.
func TestGitHubServeRemoteCanceledRequestServesCache(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
if err := p.scan(context.Background(), remote, store); err != nil {
t.Fatalf("warm scan: %v", err)
}
p.mu.Lock()
p.lastScan[remote.Name] = time.Now()
p.mu.Unlock()
ctx, cancel := context.WithCancel(context.Background())
cancel()
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-apk/x86_64/APKINDEX.tar.gz", nil).WithContext(ctx)
if !p.ServeRemote(rec, req, remote, "x86_64/APKINDEX.tar.gz", "https://x", store) {
t.Fatal("ServeRemote did not handle APKINDEX")
}
if rec.Code != http.StatusOK {
t.Fatalf("canceled request must serve cache, not error; got code=%d body=%s", rec.Code, rec.Body.String())
}
if got := readAPKIndex(t, rec.Body.Bytes()); !strings.Contains(got, "P:demo") {
t.Fatalf("expected index served from cache, got %s", got)
}
}
func TestGitHubServeRemoteRedirectRequiresReleasesRemote(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
remote.ReleasesRemote = ""
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/x", nil)
if !p.ServeRemote(rec, req, remote, demoPath, "https://x", store) {
t.Fatal("expected handled")
}
if rec.Code != http.StatusInternalServerError {
t.Fatalf("want 500 when releases_remote unset, got %d", rec.Code)
}
}
func TestGitHubScanPrunesRemovedAssets(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
store := newFakeStore()
if err := p.scan(context.Background(), fx.remote(), store); err != nil {
t.Fatalf("scan: %v", err)
}
if rows, _ := store.ListAlpineMetadataEntries(context.Background(), "acme-apk"); len(rows) != 1 {
t.Fatalf("want 1 row after first scan, got %d", len(rows))
}
delete(fx.apkBytes, "demo-1.2.3-r0.apk")
if err := p.scan(context.Background(), fx.remote(), store); err != nil {
t.Fatalf("rescan: %v", err)
}
if rows, _ := store.ListAlpineMetadataEntries(context.Background(), "acme-apk"); len(rows) != 0 {
t.Fatalf("want 0 rows after prune, got %d", len(rows))
}
}
func TestGitHubAssetPatternFilter(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.apkBytes["other-9-r0.apk"] = testsupport.MinimalApk("other", "9-r0", "aarch64")
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
remote.Patterns = []string{`^demo-.*\.apk$`}
if err := p.scan(context.Background(), remote, store); err != nil {
t.Fatalf("scan: %v", err)
}
rows, _ := store.ListAlpineMetadataEntries(context.Background(), "acme-apk")
if len(rows) != 1 || rows[0].Name != "demo" {
t.Fatalf("pattern filter failed, rows=%+v", rows)
}
}
// Multi-arch: each asset's index record lands under its own arch bucket.
func TestGitHubServeRemotePerArchGrouping(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.apkBytes["demo-1.2.3-r0-aarch64.apk"] = testsupport.MinimalApk("demo", "1.2.3-r0", "aarch64")
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
const proxyBase = "https://x"
if err := p.scan(context.Background(), remote, store); err != nil {
t.Fatalf("scan: %v", err)
}
serve := func(arch string) string {
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/x", nil)
if !p.ServeRemote(rec, req, remote, arch+"/APKINDEX.tar.gz", proxyBase, store) {
t.Fatalf("ServeRemote did not handle %s", arch)
}
return readAPKIndex(t, rec.Body.Bytes())
}
x86 := serve("x86_64")
if !strings.Contains(x86, "A:x86_64") || strings.Contains(x86, "A:aarch64") {
t.Fatalf("x86_64 index leaked another arch: %s", x86)
}
arm := serve("aarch64")
if !strings.Contains(arm, "A:aarch64") || strings.Contains(arm, "A:x86_64") {
t.Fatalf("aarch64 index leaked another arch: %s", arm)
}
}
func readAPKIndex(t *testing.T, gzBytes []byte) string {
t.Helper()
gz, err := gzip.NewReader(bytes.NewReader(gzBytes))
if err != nil {
t.Fatalf("gzip: %v", err)
}
tr := tar.NewReader(gz)
for {
hdr, err := tr.Next()
if err != nil {
t.Fatal("APKINDEX member missing from tar.gz")
}
if strings.TrimPrefix(hdr.Name, "./") == "APKINDEX" {
body, err := io.ReadAll(tr)
if err != nil {
t.Fatalf("read APKINDEX: %v", err)
}
return string(body)
}
}
}
+238
View File
@@ -0,0 +1,238 @@
package alpine
import (
"context"
"crypto/rand"
"encoding/hex"
"log/slog"
"os"
"sync"
"time"
"golang.org/x/time/rate"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
const (
syncLeaseDuration = 15 * time.Minute
defaultSyncFreshness = 5 * time.Minute
jobQueueDepth = 256
)
// SyncStore is the persistence surface the alpine syncer needs: the metadata
// cache it primes plus the shared sync-state coordination (remote enumeration
// and the per-remote lease). *database.DB satisfies it.
type SyncStore interface {
provider.RemoteMetadataStore
ListGitHubAlpineRemotes(ctx context.Context) ([]models.Remote, error)
ClaimGitHubAlpineSyncLease(ctx context.Context, remoteName, owner string, freshness, lease time.Duration) (claimed bool, etag string, err error)
ReleaseGitHubAlpineSyncLease(ctx context.Context, remoteName, owner, etag string, syncedAt time.Time) error
}
// SyncConfig tunes the shared syncer. Zero values fall back to safe defaults.
type SyncConfig struct {
RatePerSec float64
Burst int
Workers int
PollInterval time.Duration
}
type syncJob struct {
remote models.Remote
prime bool
}
// Syncer is the single per-process background worker that keeps every
// github_alpine remote's derived metadata fresh. It owns a deduped work queue, a
// pool of workers, and a global token-bucket rate limiter shared across all
// remotes and bound onto the github_alpine provider. Periodic checks are gated by
// a shared DB lease so, across replicas, only one performs each scan.
type Syncer struct {
store SyncStore
prov *GitHubProvider
limiter *rate.Limiter
cfg SyncConfig
owner string
jobs chan syncJob
mu sync.Mutex
active map[string]bool
}
// NewSyncer builds the syncer bound to the process-wide github_alpine provider
// singleton. Call Run to start it.
func NewSyncer(store SyncStore, cfg SyncConfig) *Syncer {
return newSyncer(store, gitHubProvider, cfg)
}
func newSyncer(store SyncStore, prov *GitHubProvider, cfg SyncConfig) *Syncer {
if cfg.RatePerSec <= 0 {
cfg.RatePerSec = 1
}
if cfg.Burst <= 0 {
cfg.Burst = 5
}
if cfg.Workers <= 0 {
cfg.Workers = 3
}
if cfg.PollInterval <= 0 {
cfg.PollInterval = 60 * time.Second
}
lim := rate.NewLimiter(rate.Limit(cfg.RatePerSec), cfg.Burst)
s := &Syncer{
store: store,
prov: prov,
limiter: lim,
cfg: cfg,
owner: leaseOwner(),
jobs: make(chan syncJob, jobQueueDepth),
active: map[string]bool{},
}
prov.limiter = lim
prov.syncer = s
return s
}
// Run starts the worker pool and the periodic scheduler and blocks until ctx is
// canceled, at which point it drains in-flight scans and returns.
func (s *Syncer) Run(ctx context.Context) {
slog.Info("github_alpine syncer started",
"rate_per_sec", s.cfg.RatePerSec, "burst", s.cfg.Burst,
"workers", s.cfg.Workers, "poll_interval", s.cfg.PollInterval, "owner", s.owner)
var wg sync.WaitGroup
for i := 0; i < s.cfg.Workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
s.worker(ctx)
}()
}
ticker := time.NewTicker(s.cfg.PollInterval)
defer ticker.Stop()
s.schedule(ctx)
for {
select {
case <-ctx.Done():
wg.Wait()
slog.Info("github_alpine syncer stopped")
return
case <-ticker.C:
s.schedule(ctx)
}
}
}
// schedule enqueues a periodic check for every github_alpine remote. The DB lease
// enforces the per-remote mutable_ttl cadence and cross-replica coordination.
func (s *Syncer) schedule(ctx context.Context) {
remotes, err := s.store.ListGitHubAlpineRemotes(ctx)
if err != nil {
slog.Error("github_alpine syncer: list remotes", "error", err)
return
}
for _, r := range remotes {
s.enqueue(r, false)
}
}
// EnqueuePrime queues an immediate background prime for a freshly created remote.
func (s *Syncer) EnqueuePrime(remote models.Remote) {
if s == nil {
return
}
s.enqueue(remote, true)
}
// enqueue adds a job unless the remote is already queued or in-flight, coalescing
// duplicate requests down to one scan. It never blocks.
func (s *Syncer) enqueue(remote models.Remote, prime bool) {
s.mu.Lock()
if s.active[remote.Name] {
s.mu.Unlock()
return
}
s.active[remote.Name] = true
s.mu.Unlock()
select {
case s.jobs <- syncJob{remote: remote, prime: prime}:
default:
s.mu.Lock()
delete(s.active, remote.Name)
s.mu.Unlock()
}
}
func (s *Syncer) worker(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case job := <-s.jobs:
s.process(ctx, job)
}
}
}
// process claims the shared lease and, if won, runs an incremental scan. Losing
// the claim (another replica scanning, or not yet due) is a no-op.
func (s *Syncer) process(ctx context.Context, job syncJob) {
defer func() {
s.mu.Lock()
delete(s.active, job.remote.Name)
s.mu.Unlock()
}()
freshness := time.Duration(job.remote.MutableTTL) * time.Second
if freshness <= 0 {
freshness = defaultSyncFreshness
}
if job.prime {
freshness = 0
}
claimed, etag, err := s.store.ClaimGitHubAlpineSyncLease(ctx, job.remote.Name, s.owner, freshness, syncLeaseDuration)
if err != nil {
slog.Error("github_alpine syncer: claim lease", "remote", job.remote.Name, "error", err)
return
}
if !claimed {
return
}
scanCtx, cancel := context.WithTimeout(ctx, s.prov.scanTimeout)
defer cancel()
newEtag, changed, scanErr := s.prov.scanWithState(scanCtx, job.remote, s.store, etag)
releaseEtag := etag
if scanErr == nil {
releaseEtag = newEtag
} else {
slog.Error("github_alpine syncer: scan failed", "remote", job.remote.Name, "error", scanErr)
}
relCtx, relCancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
defer relCancel()
if err := s.store.ReleaseGitHubAlpineSyncLease(relCtx, job.remote.Name, s.owner, releaseEtag, time.Now()); err != nil {
slog.Warn("github_alpine syncer: release lease", "remote", job.remote.Name, "error", err)
}
if scanErr == nil && changed {
slog.Info("github_alpine syncer: refreshed", "remote", job.remote.Name, "prime", job.prime)
}
}
// leaseOwner is a per-replica identity for the lease: hostname plus a random
// suffix so restarts and colocated replicas never collide.
func leaseOwner() string {
host, _ := os.Hostname()
var b [6]byte
_, _ = rand.Read(b[:])
return host + "-" + hex.EncodeToString(b[:])
}
+300
View File
@@ -0,0 +1,300 @@
package alpine
import (
"context"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"golang.org/x/time/rate"
"git.unkin.net/unkin/artifactapi/internal/provider"
"git.unkin.net/unkin/artifactapi/internal/testsupport"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// fakeSyncStore is an in-memory SyncStore: the metadata cache (via the embedded
// fakeStore) plus the shared sync-state lease, whose claim mirrors the atomic
// semantics of the real SQL (recency gate AND no live lease).
type fakeSyncStore struct {
*fakeStore
mu sync.Mutex
remotes []models.Remote
leaseOwner map[string]string
leaseExp map[string]time.Time
lastSynced map[string]time.Time
etags map[string]string
}
func newFakeSyncStore() *fakeSyncStore {
return &fakeSyncStore{
fakeStore: newFakeStore(),
leaseOwner: map[string]string{},
leaseExp: map[string]time.Time{},
lastSynced: map[string]time.Time{},
etags: map[string]string{},
}
}
func (f *fakeSyncStore) ListGitHubAlpineRemotes(_ context.Context) ([]models.Remote, error) {
f.mu.Lock()
defer f.mu.Unlock()
return append([]models.Remote(nil), f.remotes...), nil
}
func (f *fakeSyncStore) ClaimGitHubAlpineSyncLease(_ context.Context, name, owner string, freshness, lease time.Duration) (bool, string, error) {
f.mu.Lock()
defer f.mu.Unlock()
now := time.Now()
ls, hasLS := f.lastSynced[name]
exp, hasExp := f.leaseExp[name]
freshOK := !hasLS || now.Sub(ls) >= freshness
leaseOK := !hasExp || exp.Before(now)
if freshOK && leaseOK {
f.leaseOwner[name] = owner
f.leaseExp[name] = now.Add(lease)
return true, f.etags[name], nil
}
return false, "", nil
}
func (f *fakeSyncStore) ReleaseGitHubAlpineSyncLease(_ context.Context, name, owner, etag string, syncedAt time.Time) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.leaseOwner[name] != owner {
return nil
}
f.lastSynced[name] = syncedAt
f.etags[name] = etag
delete(f.leaseOwner, name)
delete(f.leaseExp, name)
return nil
}
func testSyncConfig() SyncConfig {
return SyncConfig{RatePerSec: 1000, Burst: 100, Workers: 1, PollInterval: time.Hour}
}
// (a) A 304 conditional response must derive nothing: no asset fetches and
// changed=false, so an unchanged repo is nearly free.
func TestSyncerConditionalNotModifiedSkipsDerive(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
etag1, changed, err := p.scanWithState(context.Background(), remote, store, "")
if err != nil {
t.Fatalf("first scan: %v", err)
}
if !changed || etag1 != `"v1"` {
t.Fatalf("first scan changed=%v etag=%q, want true and \"v1\"", changed, etag1)
}
priorRange := fx.rangeHit["demo-1.2.3-r0.apk"]
if priorRange == 0 {
t.Fatal("first scan should have fetched the asset .PKGINFO")
}
etag2, changed2, err := p.scanWithState(context.Background(), remote, store, etag1)
if err != nil {
t.Fatalf("second scan: %v", err)
}
if changed2 {
t.Fatal("304 scan must report changed=false")
}
if etag2 != etag1 {
t.Fatalf("etag changed across 304: %q -> %q", etag1, etag2)
}
if fx.notModHit != 1 {
t.Fatalf("want exactly one 304 releases response, got %d", fx.notModHit)
}
if got := fx.rangeHit["demo-1.2.3-r0.apk"]; got != priorRange {
t.Fatalf("304 scan re-fetched asset .PKGINFO: %d -> %d", priorRange, got)
}
}
// (b) On a real change, only the newly added asset is derived.
func TestSyncerIncrementalDerivesOnlyNewAsset(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
p := newTestProvider()
store := newFakeStore()
remote := fx.remote()
if _, _, err := p.scanWithState(context.Background(), remote, store, ""); err != nil {
t.Fatalf("first scan: %v", err)
}
demoRange := fx.rangeHit["demo-1.2.3-r0.apk"]
fx.apkBytes["other-9-r0.apk"] = testsupport.MinimalApk("other", "9-r0", "aarch64")
fx.etag = `"v2"`
if _, changed, err := p.scanWithState(context.Background(), remote, store, `"v1"`); err != nil || !changed {
t.Fatalf("second scan changed=%v err=%v", changed, err)
}
rows, _ := store.ListAlpineMetadataEntries(context.Background(), remote.Name)
if len(rows) != 2 {
t.Fatalf("want 2 cached rows after incremental derive, got %d", len(rows))
}
if got := fx.rangeHit["demo-1.2.3-r0.apk"]; got != demoRange {
t.Fatalf("already-cached asset was re-fetched: %d -> %d", demoRange, got)
}
if fx.rangeHit["other-9-r0.apk"] == 0 {
t.Fatal("newly added asset was not derived")
}
}
// (c) The shared limiter caps the request rate.
func TestRateLimiterCapsRequestRate(t *testing.T) {
fx := newGitHubFixture(t, true)
p := newTestProvider()
p.limiter = rate.NewLimiter(rate.Every(120*time.Millisecond), 1)
remote := fx.remote()
start := time.Now()
for i := 0; i < 3; i++ {
if _, _, _, err := p.fetchReleases(context.Background(), remote, ""); err != nil {
t.Fatalf("fetchReleases %d: %v", i, err)
}
}
if elapsed := time.Since(start); elapsed < 200*time.Millisecond {
t.Fatalf("rate limiter did not throttle: 3 calls took %v, want >= 200ms", elapsed)
}
}
// (d) Concurrent enqueues for the same remote coalesce to a single queued job.
func TestSyncerEnqueueDedup(t *testing.T) {
store := newFakeSyncStore()
p := newTestProvider()
s := newSyncer(store, p, testSyncConfig())
remote := models.Remote{Name: "acme-apk", PackageType: models.PackageGitHubAlpine, MutableTTL: 3600}
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go func() { defer wg.Done(); s.enqueue(remote, false) }()
}
wg.Wait()
if got := len(s.jobs); got != 1 {
t.Fatalf("want exactly 1 coalesced job, got %d", got)
}
}
// (e) Prime-on-create enqueues a prime job.
func TestSyncerEnqueuePrime(t *testing.T) {
store := newFakeSyncStore()
p := newTestProvider()
s := newSyncer(store, p, testSyncConfig())
remote := models.Remote{Name: "acme-apk", PackageType: models.PackageGitHubAlpine, MutableTTL: 3600}
s.EnqueuePrime(remote)
select {
case job := <-s.jobs:
if !job.prime || job.remote.Name != "acme-apk" {
t.Fatalf("bad prime job: %+v", job)
}
default:
t.Fatal("EnqueuePrime did not enqueue a job")
}
}
// (f) A held lease prevents a second replica from scanning.
func TestSyncerLeasePreventsSecondReplica(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
store := newFakeSyncStore()
p := newTestProvider()
s := newSyncer(store, p, testSyncConfig())
remote := fx.remote()
claimed, _, err := store.ClaimGitHubAlpineSyncLease(context.Background(), remote.Name, "replica-1", time.Duration(remote.MutableTTL)*time.Second, syncLeaseDuration)
if err != nil || !claimed {
t.Fatalf("replica-1 claim: claimed=%v err=%v", claimed, err)
}
s.process(context.Background(), syncJob{remote: remote})
if fx.releasesHit != 0 {
t.Fatalf("second replica scanned while lease held: %d releases calls", fx.releasesHit)
}
if rows, _ := store.ListAlpineMetadataEntries(context.Background(), remote.Name); len(rows) != 0 {
t.Fatalf("second replica derived metadata while lease held: %d rows", len(rows))
}
}
// With the syncer wired and the cache empty, an index request enqueues a prime
// and returns a retryable 503 when it has not landed within the cold wait.
func TestServeRemoteColdStartReturns503(t *testing.T) {
fx := newGitHubFixture(t, true)
store := newFakeSyncStore()
p := newTestProvider()
p.coldWait = 300 * time.Millisecond
_ = newSyncer(store, p, testSyncConfig()) // binds p.syncer, but no workers running
remote := fx.remote()
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-apk/x86_64/APKINDEX.tar.gz", nil)
if !p.ServeRemote(rec, req, remote, "x86_64/APKINDEX.tar.gz", "https://x", store) {
t.Fatal("ServeRemote did not handle APKINDEX")
}
if rec.Code != http.StatusServiceUnavailable {
t.Fatalf("cold empty cache must return 503, got %d", rec.Code)
}
if rec.Header().Get("Retry-After") == "" {
t.Fatal("503 should carry Retry-After")
}
if got := len(p.syncer.jobs); got != 1 {
t.Fatalf("cold start did not enqueue a prime, jobs=%d", got)
}
}
// With the cache warm, the same request serves the index immediately (no 503).
func TestServeRemoteWarmCacheServesImmediately(t *testing.T) {
fx := newGitHubFixture(t, true)
store := newFakeSyncStore()
p := newTestProvider()
_ = newSyncer(store, p, testSyncConfig())
remote := fx.remote()
if err := p.scan(context.Background(), remote, store); err != nil {
t.Fatalf("warm scan: %v", err)
}
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/api/v1/remote/acme-apk/x86_64/APKINDEX.tar.gz", nil)
if !p.ServeRemote(rec, req, remote, "x86_64/APKINDEX.tar.gz", "https://x", store) {
t.Fatal("ServeRemote did not handle APKINDEX")
}
if rec.Code != http.StatusOK {
t.Fatalf("warm cache must serve 200, got %d body=%s", rec.Code, rec.Body.String())
}
}
// A prime job (freshness 0) runs even right after a sync; a periodic job at the
// same moment is gated by the recency window.
func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
store := newFakeSyncStore()
p := newTestProvider()
s := newSyncer(store, p, testSyncConfig())
remote := fx.remote()
var _ provider.RemoteMetadataStore = store
s.process(context.Background(), syncJob{remote: remote, prime: true})
if rows, _ := store.ListAlpineMetadataEntries(context.Background(), remote.Name); len(rows) != 1 {
t.Fatalf("prime did not derive: %d rows", len(rows))
}
releasesAfterPrime := fx.releasesHit
s.process(context.Background(), syncJob{remote: remote, prime: false})
if fx.releasesHit != releasesAfterPrime {
t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit)
}
}
+14 -3
View File
@@ -20,7 +20,7 @@ import (
"git.unkin.net/unkin/artifactapi/internal/database"
"git.unkin.net/unkin/artifactapi/internal/gc"
"git.unkin.net/unkin/artifactapi/internal/githubauth"
_ "git.unkin.net/unkin/artifactapi/internal/provider/alpine"
"git.unkin.net/unkin/artifactapi/internal/provider/alpine"
"git.unkin.net/unkin/artifactapi/internal/provider/deb"
_ "git.unkin.net/unkin/artifactapi/internal/provider/docker"
_ "git.unkin.net/unkin/artifactapi/internal/provider/generic"
@@ -52,6 +52,7 @@ type Server struct {
gc *gc.Collector
syncer *rpm.Syncer
debSyncer *deb.Syncer
alpineSyncer *alpine.Syncer
}
func New(cfg *config.Config, version string) (*Server, error) {
@@ -105,6 +106,12 @@ func New(cfg *config.Config, version string) (*Server, error) {
Workers: cfg.GitHubSyncWorkers,
PollInterval: time.Duration(cfg.GitHubSyncPollInterval) * time.Second,
})
alpineSyncer := alpine.NewSyncer(db, alpine.SyncConfig{
RatePerSec: cfg.GitHubSyncRatePerSec,
Burst: cfg.GitHubSyncBurst,
Workers: cfg.GitHubSyncWorkers,
PollInterval: time.Duration(cfg.GitHubSyncPollInterval) * time.Second,
})
// The terraform registry signs with a GPG key. A configured file wins (BYO
// key); otherwise artifactapi generates one on first start and persists it in
@@ -138,6 +145,7 @@ func New(cfg *config.Config, version string) (*Server, error) {
gc: collector,
syncer: syncer,
debSyncer: debSyncer,
alpineSyncer: alpineSyncer,
}
s.router = s.routes()
@@ -168,8 +176,9 @@ func (s *Server) routes() chi.Router {
r.Mount("/v2", proxyHandler.DockerV2Routes())
remotesHandler := v2.NewRemotesHandler(s.db, map[models.PackageType]v2.Primer{
models.PackageGitHubRPM: s.syncer,
models.PackageGitHubDeb: s.debSyncer,
models.PackageGitHubRPM: s.syncer,
models.PackageGitHubDeb: s.debSyncer,
models.PackageGitHubAlpine: s.alpineSyncer,
})
virtualsHandler := v2.NewVirtualsHandler(s.db)
healthHandler := v2.NewHealthHandler(s.db, s.cache, s.store)
@@ -239,6 +248,7 @@ func (s *Server) Run(ctx context.Context) error {
go s.gc.Run(ctx)
go s.syncer.Run(ctx)
go s.debSyncer.Run(ctx)
go s.alpineSyncer.Run(ctx)
httpServer := s.newHTTPServer()
@@ -261,6 +271,7 @@ func (s *Server) RunOnListener(ctx context.Context, ln net.Listener) error {
go s.gc.Run(ctx)
go s.syncer.Run(ctx)
go s.debSyncer.Run(ctx)
go s.alpineSyncer.Run(ctx)
httpServer := s.newHTTPServer()
+28 -26
View File
@@ -5,35 +5,37 @@ import "fmt"
type PackageType string
const (
PackageGeneric PackageType = "generic"
PackageDocker PackageType = "docker"
PackageHelm PackageType = "helm"
PackagePyPI PackageType = "pypi"
PackageNPM PackageType = "npm"
PackageRPM PackageType = "rpm"
PackageDeb PackageType = "deb"
PackageAlpine PackageType = "alpine"
PackagePuppet PackageType = "puppet"
PackageTerraform PackageType = "terraform"
PackageGoProxy PackageType = "goproxy"
PackageGitHubRPM PackageType = "github_rpm"
PackageGitHubDeb PackageType = "github_deb"
PackageGeneric PackageType = "generic"
PackageDocker PackageType = "docker"
PackageHelm PackageType = "helm"
PackagePyPI PackageType = "pypi"
PackageNPM PackageType = "npm"
PackageRPM PackageType = "rpm"
PackageDeb PackageType = "deb"
PackageAlpine PackageType = "alpine"
PackagePuppet PackageType = "puppet"
PackageTerraform PackageType = "terraform"
PackageGoProxy PackageType = "goproxy"
PackageGitHubRPM PackageType = "github_rpm"
PackageGitHubDeb PackageType = "github_deb"
PackageGitHubAlpine PackageType = "github_alpine"
)
var validPackageTypes = map[PackageType]bool{
PackageGeneric: true,
PackageDocker: true,
PackageHelm: true,
PackagePyPI: true,
PackageNPM: true,
PackageRPM: true,
PackageDeb: true,
PackageAlpine: true,
PackagePuppet: true,
PackageTerraform: true,
PackageGoProxy: true,
PackageGitHubRPM: true,
PackageGitHubDeb: true,
PackageGeneric: true,
PackageDocker: true,
PackageHelm: true,
PackagePyPI: true,
PackageNPM: true,
PackageRPM: true,
PackageDeb: true,
PackageAlpine: true,
PackagePuppet: true,
PackageTerraform: true,
PackageGoProxy: true,
PackageGitHubRPM: true,
PackageGitHubDeb: true,
PackageGitHubAlpine: true,
}
func (p PackageType) Valid() bool {
+2
View File
@@ -19,6 +19,8 @@ func TestPackageTypeValid(t *testing.T) {
models.PackageTerraform,
models.PackageGoProxy,
models.PackageGitHubRPM,
models.PackageGitHubDeb,
models.PackageGitHubAlpine,
}
for _, pt := range valid {
if !pt.Valid() {