Guard cargo publishes against name and version collisions

- serialise publishes per crate with an advisory lock in one transaction
- add a unique index on normalised name and version without build metadata
- validate versions as strict SemVer 2.0 and reject an empty .crate
- renumber the migration to 0005

e2e-docker/cargo_local_test.go

@@ -19,18 +19,28 @@ func TestLocalCargoRegistry(t *testing.T) {
 
 	meta := `{"name":"e2e-crate","vers":"0.1.0","deps":[],"features":{}}`
 	crate := []byte("e2e crate bytes")
-	var body []byte
-	body = binary.LittleEndian.AppendUint32(body, uint32(len(meta)))
-	body = append(body, meta...)
-	body = binary.LittleEndian.AppendUint32(body, uint32(len(crate)))
-	body = append(body, crate...)
-
-	if resp, b := doRequest(t, http.MethodPut, base+"/api/v1/crates/new", body, "application/octet-stream"); resp.StatusCode != http.StatusOK {
-		t.Fatalf("publish: status %d: %s", resp.StatusCode, b)
+	publish := func(meta string) *http.Response {
+		var body []byte
+		body = binary.LittleEndian.AppendUint32(body, uint32(len(meta)))
+		body = append(body, meta...)
+		body = binary.LittleEndian.AppendUint32(body, uint32(len(crate)))
+		body = append(body, crate...)
+		resp, _ := doRequest(t, http.MethodPut, base+"/api/v1/crates/new", body, "application/octet-stream")
+		return resp
 	}
-	if resp, _ := doRequest(t, http.MethodPut, base+"/api/v1/crates/new", body, "application/octet-stream"); resp.StatusCode != http.StatusConflict {
+
+	if resp := publish(meta); resp.StatusCode != http.StatusOK {
+		t.Fatalf("publish: status %d", resp.StatusCode)
+	}
+	if resp := publish(meta); resp.StatusCode != http.StatusConflict {
 		t.Fatalf("republish: status %d, want 409", resp.StatusCode)
 	}
+	if resp := publish(`{"name":"e2e-crate","vers":"0.1.0+other","deps":[],"features":{}}`); resp.StatusCode != http.StatusConflict {
+		t.Fatalf("build-metadata republish: status %d, want 409", resp.StatusCode)
+	}
+	if resp := publish(`{"name":"E2E_crate","vers":"0.2.0","deps":[],"features":{}}`); resp.StatusCode != http.StatusBadRequest {
+		t.Fatalf("name collision publish: status %d, want 400", resp.StatusCode)
+	}
 
 	resp, b := doRequest(t, http.MethodGet, base+"/config.json", nil, "")
 	if resp.StatusCode != http.StatusOK || !strings.Contains(string(b), `"api":"`) {

internal/api/v1/cargo.go

@@ -73,22 +73,6 @@ func (h *ProxyHandler) cargoPublish(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	existing, err := h.db.ListCargoIndexEntries(ctx, remote.Name, meta.Name)
-	if err != nil {
-		cargoError(w, http.StatusInternalServerError, "%v", err)
-		return
-	}
-	for _, e := range existing {
-		if e.Name != meta.Name {
-			cargoError(w, http.StatusBadRequest, "crate %q already exists as %q", meta.Name, e.Name)
-			return
-		}
-		if e.Version == meta.Vers {
-			cargoError(w, http.StatusConflict, "crate version `%s@%s` already exists", meta.Name, meta.Vers)
-			return
-		}
-	}
-
 	lineJSON, err := json.Marshal(line)
 	if err != nil {
 		cargoError(w, http.StatusInternalServerError, "%v", err)
@@ -117,7 +101,14 @@ func (h *ProxyHandler) cargoPublish(w http.ResponseWriter, r *http.Request) {
 	if err := h.db.InsertCargoIndexEntry(ctx, entry); err != nil {
 		// Drop the file so a retry is not rejected as a duplicate.
 		_ = h.db.DeleteLocalFile(ctx, remote.Name, path)
-		cargoError(w, http.StatusInternalServerError, "record index: %v", err)
+		switch {
+		case errors.Is(err, database.ErrCargoNameTaken):
+			cargoError(w, http.StatusBadRequest, "%v", err)
+		case errors.Is(err, database.ErrAlreadyExists):
+			cargoError(w, http.StatusConflict, "crate version `%s@%s` already exists", meta.Name, meta.Vers)
+		default:
+			cargoError(w, http.StatusInternalServerError, "record index: %v", err)
+		}
 		return
 	}
 

internal/database/cargo_index.go

@@ -2,17 +2,72 @@ package database
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"strings"
 
+	"github.com/jackc/pgx/v5/pgconn"
+
 	"git.unkin.net/unkin/artifactapi/internal/provider"
 )
 
+var ErrCargoNameTaken = errors.New("crate name already taken")
+
+func cargoNormName(name string) string {
+	return strings.ReplaceAll(strings.ToLower(name), "-", "_")
+}
+
+// InsertCargoIndexEntry rejects a name that differs from an existing crate only
+// by case or -/_ (ErrCargoNameTaken) and a version equal up to build metadata
+// (ErrAlreadyExists). A per-crate advisory lock serialises concurrent publishes.
 func (db *DB) InsertCargoIndexEntry(ctx context.Context, e *provider.CargoIndexEntry) error {
-	_, err := db.Pool.Exec(ctx, `
+	tx, err := db.Pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+
+	norm := cargoNormName(e.Name)
+	if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('cargo:' || $1 || '/' || $2))`, e.RepoName, norm); err != nil {
+		return err
+	}
+	rows, err := tx.Query(ctx, `
+		SELECT name, version FROM cargo_index
+		WHERE repo_name = $1 AND replace(lower(name), '-', '_') = $2
+	`, e.RepoName, norm)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	base, _, _ := strings.Cut(e.Version, "+")
+	for rows.Next() {
+		var name, version string
+		if err := rows.Scan(&name, &version); err != nil {
+			return err
+		}
+		if name != e.Name {
+			return fmt.Errorf("%w: %q exists as %q", ErrCargoNameTaken, e.Name, name)
+		}
+		if v, _, _ := strings.Cut(version, "+"); v == base {
+			return fmt.Errorf("%w: %s@%s", ErrAlreadyExists, name, version)
+		}
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	rows.Close()
+
+	if _, err := tx.Exec(ctx, `
 		INSERT INTO cargo_index (repo_name, file_path, name, version, line, yanked)
 		VALUES ($1, $2, $3, $4, $5, $6)
-	`, e.RepoName, e.FilePath, e.Name, e.Version, e.Line, e.Yanked)
-	return err
+	`, e.RepoName, e.FilePath, e.Name, e.Version, e.Line, e.Yanked); err != nil {
+		var pgErr *pgconn.PgError
+		if errors.As(err, &pgErr) && pgErr.Code == "23505" {
+			return fmt.Errorf("%w: %s@%s", ErrAlreadyExists, e.Name, e.Version)
+		}
+		return err
+	}
+	return tx.Commit(ctx)
 }
 
 func (db *DB) ListCargoIndexEntries(ctx context.Context, repoName, name string) ([]provider.CargoIndexEntry, error) {

internal/database/database_test.go

@@ -2,7 +2,10 @@ package database
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"os"
+	"sync"
 	"testing"
 	"time"
 
@@ -460,3 +463,80 @@ func TestCargoIndex(t *testing.T) {
 		t.Fatalf("after delete: %+v", entries)
 	}
 }
+
+func TestCargoIndexPublishConflicts(t *testing.T) {
+	requireDB(t)
+	entry := func(name, version string) *provider.CargoIndexEntry {
+		return &provider.CargoIndexEntry{
+			RepoName: "cargo-conflict", FilePath: "crates/" + name + "/" + name + "-" + version + ".crate",
+			Name: name, Version: version, Line: "{}",
+		}
+	}
+	if err := testDB.InsertCargoIndexEntry(ctx(), entry("foo-bar", "1.0.0+a")); err != nil {
+		t.Fatal(err)
+	}
+	for _, tc := range []struct {
+		name, version string
+		want          error
+	}{
+		{"foo-bar", "1.0.0+b", ErrAlreadyExists},
+		{"foo-bar", "1.0.0", ErrAlreadyExists},
+		{"Foo-Bar", "2.0.0", ErrCargoNameTaken},
+		{"foo_bar", "2.0.0", ErrCargoNameTaken},
+	} {
+		if err := testDB.InsertCargoIndexEntry(ctx(), entry(tc.name, tc.version)); !errors.Is(err, tc.want) {
+			t.Errorf("%s@%s: err = %v, want %v", tc.name, tc.version, err, tc.want)
+		}
+	}
+	if err := testDB.InsertCargoIndexEntry(ctx(), entry("foo-bar", "1.0.1")); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestCargoIndexConcurrentPublish(t *testing.T) {
+	requireDB(t)
+	names := []string{"Race", "race", "RACE", "race", "rAce", "RaCe", "racE", "race"}
+	errs := make(chan error, len(names))
+	var wg sync.WaitGroup
+	for i, name := range names {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			errs <- testDB.InsertCargoIndexEntry(ctx(), &provider.CargoIndexEntry{
+				RepoName: "cargo-race", FilePath: fmt.Sprintf("crates/%s/%d.crate", name, i),
+				Name: name, Version: "1.0.0", Line: "{}",
+			})
+		}()
+	}
+	wg.Wait()
+	close(errs)
+	ok := 0
+	for err := range errs {
+		switch {
+		case err == nil:
+			ok++
+		case !errors.Is(err, ErrAlreadyExists) && !errors.Is(err, ErrCargoNameTaken):
+			t.Errorf("unexpected error: %v", err)
+		}
+	}
+	if ok != 1 {
+		t.Fatalf("%d concurrent publishes succeeded, want 1", ok)
+	}
+	if entries, _ := testDB.ListCargoIndexEntries(ctx(), "cargo-race", "race"); len(entries) > 1 {
+		t.Fatalf("entries = %+v", entries)
+	}
+}
+
+func TestCargoIndexUniqueIndex(t *testing.T) {
+	requireDB(t)
+	insert := func(name, version, path string) error {
+		_, err := testDB.Pool.Exec(ctx(), `INSERT INTO cargo_index (repo_name, file_path, name, version, line) VALUES ('cargo-uniq', $1, $2, $3, '{}')`, path, name, version)
+		return err
+	}
+	if err := insert("a-b", "1.0.0+x", "p1"); err != nil {
+		t.Fatal(err)
+	}
+	if err := insert("A_B", "1.0.0+y", "p2"); err == nil {
+		t.Fatal("unique index allowed a normalised duplicate")
+	}
+}

internal/provider/cargo/local.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
+	"regexp"
 	"strings"
 
 	"git.unkin.net/unkin/artifactapi/internal/provider"
@@ -105,6 +106,9 @@ func NewIndexLine(meta PublishMeta, crate []byte) (IndexLine, error) {
 	if !validVersion(meta.Vers) {
 		return IndexLine{}, fmt.Errorf("invalid version %q", meta.Vers)
 	}
+	if len(crate) == 0 {
+		return IndexLine{}, errors.New("empty .crate")
+	}
 	sum := sha256.Sum256(crate)
 	line := IndexLine{
 		Name:        meta.Name,
@@ -184,11 +188,11 @@ func validName(name string) bool {
 	return true
 }
 
+// semverRe is the SemVer 2.0.0 grammar from semver.org.
+var semverRe = regexp.MustCompile(`^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$`)
+
 func validVersion(v string) bool {
-	if v == "" || len(v) > 128 || !strings.ContainsAny(v[:1], "0123456789") {
-		return false
-	}
-	return strings.Trim(v, "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-+") == ""
+	return len(v) <= 128 && semverRe.MatchString(v)
 }
 
 // IndexPath is the sparse-index file path for a crate name.

internal/provider/cargo/local_test.go

@@ -90,7 +90,7 @@ func TestParsePublishAndIndexLine(t *testing.T) {
 }
 
 func TestIndexLineNoNewFeaturesOmitsV(t *testing.T) {
-	line, err := NewIndexLine(PublishMeta{Name: "a", Vers: "1.0.0", Features: map[string][]string{"x": nil}}, nil)
+	line, err := NewIndexLine(PublishMeta{Name: "a", Vers: "1.0.0", Features: map[string][]string{"x": nil}}, []byte("x"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -129,13 +129,27 @@ func TestNewIndexLineValidation(t *testing.T) {
 		{Name: "ok", Vers: "1.0.0/../../x"},
 		{Name: "ok", Vers: ""},
 		{Name: strings.Repeat("a", 65), Vers: "1.0.0"},
+		{Name: "ok", Vers: "1"},
+		{Name: "ok", Vers: "1.0"},
+		{Name: "ok", Vers: "1..2"},
+		{Name: "ok", Vers: "01.0.0"},
+		{Name: "ok", Vers: "1.0.0-"},
+		{Name: "ok", Vers: "1.0.0-01"},
+		{Name: "ok", Vers: "1.0.0+"},
+		{Name: "ok", Vers: "1.0.0+a..b"},
+		{Name: "ok", Vers: "v1.0.0"},
 	} {
-		if _, err := NewIndexLine(m, nil); err == nil {
+		if _, err := NewIndexLine(m, []byte("x")); err == nil {
 			t.Errorf("%q@%q accepted", m.Name, m.Vers)
 		}
 	}
-	if _, err := NewIndexLine(PublishMeta{Name: "Foo_bar-2", Vers: "1.0.0-rc.1+build.5"}, nil); err != nil {
-		t.Error(err)
+	for _, v := range []string{"0.0.0", "1.0.0-rc.1+build.5", "1.2.3-alpha.0.x-y", "10.20.30+20261017"} {
+		if _, err := NewIndexLine(PublishMeta{Name: "Foo_bar-2", Vers: v}, []byte("x")); err != nil {
+			t.Error(err)
+		}
+	}
+	if _, err := NewIndexLine(PublishMeta{Name: "ok", Vers: "1.0.0"}, nil); err == nil {
+		t.Error("empty .crate accepted")
 	}
 }
 

migrations/0005_cargo_index.sql

@@ -13,3 +13,7 @@ CREATE TABLE IF NOT EXISTS cargo_index (
 );
 
 CREATE INDEX IF NOT EXISTS idx_cargo_index_repo_name ON cargo_index(repo_name, lower(name));
+
+-- Cargo treats names equal up to case and -/_, and versions equal up to build metadata.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_cargo_index_unique_version
+	ON cargo_index(repo_name, replace(lower(name), '-', '_'), split_part(version, '+', 1));