migrate

.gitea/workflows/docker-publish.yml

@@ -36,7 +36,7 @@ jobs:
   build:
     runs-on: ubuntu-latest
     permissions:
-      contents: read
+      contents: write    # the release step below
       packages: write
 
     steps:
@@ -82,9 +82,48 @@ jobs:
             type=raw,value=${{ inputs.tag }},enable=${{ inputs.tag != '' }}
 
       - name: Build and push Docker image
+        id: build
         uses: docker/build-push-action@v5
         with:
           context: .
           push: true
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}
+
+      # Gitea's Packages tab only lists images in Gitea's own registry, so a v* tag gets a
+      # release instead, naming the image it shipped. The digest line is the one to trust:
+      # ipcr 1.7.1 can drop some of the tags pushed together (see the tag list).
+      - name: Publish the Gitea release
+        if: startsWith(github.ref, 'refs/tags/v')
+        env:
+          TOKEN: ${{ secrets.GITEA_TOKEN }}
+          API: ${{ github.server_url }}/api/v1/repos/${{ github.repository }}
+          TAG: ${{ github.ref_name }}
+          SHA: ${{ github.sha }}
+          DIGEST: ${{ steps.build.outputs.digest }}
+          TAGS: ${{ steps.meta.outputs.tags }}
+          PULL: ipcr-watch.nsl.sh/ipns/metadec.eth/${{ env.IMAGE_NAME }}
+        run: |
+          node -e '
+            const e = process.env, ver = e.TAG.replace(/^v/, "");
+            const tags = e.TAGS.split("\n").filter(Boolean).map(t => "`" + t.split(":").pop() + "`");
+            const body = [
+              "**Image** (anonymous pull):",
+              "```", `docker pull ${e.PULL}:${ver}`, "```",
+              "Pinned by digest:",
+              "```", `${e.PULL}@${e.DIGEST}`, "```",
+              `Tags pushed: ${tags.join(", ")} · commit ${e.SHA.slice(0, 7)}`,
+            ].join("\n");
+            console.log(JSON.stringify({ tag_name: e.TAG, name: e.TAG, body }));
+          ' > /tmp/release.json
+          auth="Authorization: token $TOKEN"
+          code=$(curl -s -o /tmp/existing.json -w '%{http_code}' -H "$auth" "$API/releases/tags/$TAG")
+          if [ "$code" = 200 ]; then
+            id=$(node -p 'require("/tmp/existing.json").id')
+            method=PATCH url="$API/releases/$id"
+          else
+            method=POST url="$API/releases"
+          fi
+          code=$(curl -s -o /tmp/out.json -w '%{http_code}' -X "$method" -H "$auth" \
+            -H 'Content-Type: application/json' --data @/tmp/release.json "$url")
+          case "$code" in 2*) echo "release $TAG: $method $code" ;; *) cat /tmp/out.json; exit 1 ;; esac

tools/claims-migrate/migrate.go

@@ -61,6 +61,9 @@ type stats struct {
 
 	CIDGroupRows  int
 	CIDConflicts  int // cids/<c> on a rid while cid:<c> names another rid
+	CIDShared     int // cids/<c> (no cid: row) already kept by an earlier record
+	RecordsFolded int // records whose every cid an earlier record kept: merged into it
+	DupClaims     int // (cid, key) already claimed by the legacy signer (folded records)
 	OrphanCIDRows int // cid:<c> → rid with no file: rows
 
 	UDLMigrated  counts
@@ -74,6 +77,8 @@ type stats struct {
 	Principals    int
 	ClaimsWritten int
 	Elapsed       time.Duration
+
+	canon map[string]string // rid → the canonical cid its claims sit on
 }
 
 func newStats() *stats {
@@ -81,6 +86,7 @@ func newStats() *stats {
 		WatcherFacts: counts{}, RefusedFields: counts{},
 		UDLMigrated: counts{}, UDLDropped: counts{},
 		Scratch: counts{}, Derived: counts{},
+		canon: map[string]string{},
 	}
 }
 
@@ -239,6 +245,7 @@ func migrate(o options) (st *stats, err error) {
 	st.CIDGroupRows = len(groups)
 	st.Principals = len(w.pids)
 	st.ClaimsWritten = int(w.seq)
+	st.DupClaims = w.dups
 	if !o.dryRun {
 		if err = w.finish(groups); err != nil {
 			return nil, err
@@ -439,14 +446,40 @@ func emitRecord(o options, st *stats, legacy *identity.Identity, index map[strin
 		st.RecordsNoCID++
 		return nil
 	}
-	canonical := usable[0]
-	for _, c := range usable[1:] {
-		canonical = cid.Better(canonical, c)
-	}
+	// A cid with no cid: index row can sit in two records' cids/* (the same
+	// content stored twice). The first record keeps it; a record left with none
+	// folds into the record owning its first shared cid, and its fields become
+	// claims there (the legacy signer's first claim per (cid, key) wins).
+	owner, shared := "", ""
+	var own []string
 	for _, c := range usable {
-		groups[c] = rid
+		if r, ok := groups[c]; ok && r != rid {
+			st.CIDShared++
+			if owner == "" {
+				owner, shared = r, c
+			}
+			continue
+		}
+		own = append(own, c)
+	}
+	var canonical string
+	if len(own) == 0 {
+		st.RecordsFolded++
+		rid = owner
+		if canonical = st.canon[owner]; canonical == "" {
+			canonical = shared
+		}
+	} else {
+		canonical = own[0]
+		for _, c := range own[1:] {
+			canonical = cid.Better(canonical, c)
+		}
+		for _, c := range own {
+			groups[c] = rid
+		}
+		st.canon[rid] = canonical
+		st.Rids++
 	}
-	st.Rids++
 	emitted := 0
 	mk := func(key, value string) error {
 		emitted++
@@ -685,6 +718,7 @@ type writer struct {
 	db   *sql.DB
 	pids map[string]int64
 	seq  int64
+	dups int
 }
 
 func (w *writer) write(batch []claim.Claim) error {
@@ -702,7 +736,7 @@ func (w *writer) write(batch []claim.Claim) error {
 		return err
 	}
 	defer tx.Rollback()
-	ins, err := tx.Prepare(`INSERT INTO claim(cid, key, pid, value, ts, sig, seq) VALUES(?, ?, ?, ?, ?, ?, ?)`)
+	ins, err := tx.Prepare(`INSERT OR IGNORE INTO claim(cid, key, pid, value, ts, sig, seq) VALUES(?, ?, ?, ?, ?, ?, ?)`)
 	if err != nil {
 		return err
 	}
@@ -716,9 +750,14 @@ func (w *writer) write(batch []claim.Claim) error {
 			w.pids[cl.UID] = pid
 		}
 		w.seq++
-		if _, err := ins.Exec(cl.CID, cl.Key, pid, valueArg(cl.Value), int64(cl.TS), cl.Sig, w.seq); err != nil {
+		res, err := ins.Exec(cl.CID, cl.Key, pid, valueArg(cl.Value), int64(cl.TS), cl.Sig, w.seq)
+		if err != nil {
 			return fmt.Errorf("claim (%s, %s, %s): %w", cl.CID, cl.Key, cl.UID, err)
 		}
+		if n, _ := res.RowsAffected(); n == 0 { // a folded record's duplicate
+			w.seq--
+			w.dups++
+		}
 	}
 	return tx.Commit()
 }
@@ -875,6 +914,8 @@ func printSummary(w io.Writer, st *stats, dryRun bool) {
 	fmt.Fprintf(w, "principals               %d\n", st.Principals)
 	fmt.Fprintf(w, "cid_group rows           %d  (%d minted for UDL-only cids)\n", st.CIDGroupRows, st.GroupsMinted)
 	fmt.Fprintf(w, "cid conflicts            %d  (cids/<c> on a rid while cid:<c> names another; index wins)\n", st.CIDConflicts)
+	fmt.Fprintf(w, "shared cids              %d  (cids/<c> kept by an earlier record; %d records folded into it)\n", st.CIDShared, st.RecordsFolded)
+	fmt.Fprintf(w, "duplicate claims skipped %d  (folded records' (cid, key) already claimed; not counted in dry run)\n", st.DupClaims)
 	fmt.Fprintf(w, "orphan cid: rows dropped %d  (rid has no file: rows)\n", st.OrphanCIDRows)
 	fmt.Fprintf(w, "watcher facts skipped    %d\n", st.WatcherFacts.total())
 	printCounts(w, "    ", st.WatcherFacts)

tools/claims-migrate/migrate_test.go

@@ -373,3 +373,71 @@ func TestHalfFloat(t *testing.T) {
 		}
 	}
 }
+
+// Two records listing the same cid in cids/* with no cid: index row (the same
+// content stored twice): the first keeps it, a record left with no cid of its
+// own folds into it, and the duplicate (cid, key) claims are skipped.
+func TestSharedUnindexedCIDFolds(t *testing.T) {
+	dir := t.TempDir()
+	oldPath := filepath.Join(dir, "v1.db")
+	db, err := sql.Open("sqlite", "file:"+oldPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`CREATE TABLE kv(k TEXT PRIMARY KEY, v TEXT NOT NULL) WITHOUT ROWID`); err != nil {
+		t.Fatal(err)
+	}
+	for k, v := range map[string]string{
+		"meta-core:schema-version":         "2",
+		"file:" + ridA + "/cids/" + cidSHA: "true",
+		"file:" + ridA + "/title":          "A",
+		"file:" + ridB + "/cids/" + cidSHA: "true",
+		"file:" + ridB + "/title":          "B",
+		"file:" + ridB + "/genres/Drama":   "true",
+		"file:" + ridC + "/cids/" + cidSHA: "true",
+		"file:" + ridC + "/cids/" + cidB:   "true",
+		"file:" + ridC + "/title":          "C",
+	} {
+		if _, err := db.Exec(`INSERT INTO kv(k, v) VALUES(?, ?)`, k, v); err != nil {
+			t.Fatal(err)
+		}
+	}
+	db.Close()
+	truthPath := filepath.Join(dir, "truth.db")
+	st, err := migrate(options{oldPath: oldPath, truthPath: truthPath, workers: 2})
+	if err != nil {
+		t.Fatalf("migrate: %v", err)
+	}
+	if st.Rids != 2 || st.RecordsFolded != 1 || st.CIDShared != 2 || st.DupClaims != 1 {
+		t.Errorf("rids %d folded %d shared %d dups %d", st.Rids, st.RecordsFolded, st.CIDShared, st.DupClaims)
+	}
+	out, err := sql.Open("sqlite", "file:"+truthPath+"?mode=ro")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer out.Close()
+	groups := map[string]string{}
+	rows, err := out.Query(`SELECT cid, rid FROM cid_group`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for rows.Next() {
+		var c, r string
+		if err := rows.Scan(&c, &r); err != nil {
+			t.Fatal(err)
+		}
+		groups[c] = r
+	}
+	rows.Close()
+	if groups[cidSHA] != ridA || groups[cidB] != ridC || len(groups) != 2 {
+		t.Errorf("cid_group %v", groups)
+	}
+	var n int
+	if err := out.QueryRow(`SELECT count(*) FROM claim WHERE cid = ? AND key IN ('title', 'genres/Drama')`, cidSHA).Scan(&n); err != nil || n != 2 {
+		t.Errorf("claims on the shared cid: %d (%v)", n, err)
+	}
+	var title string
+	if err := out.QueryRow(`SELECT value FROM claim WHERE cid = ? AND key = 'title'`, cidB).Scan(&title); err != nil || title != "C" {
+		t.Errorf("record C title on its own cid: %q (%v)", title, err)
+	}
+}