pages/internal/store/maintenance_test.go

package store

import (
	"context"
	"errors"
	"strconv"
	"testing"
	"time"

	"github.com/iceBear67/simplepages/internal/cas"
)

// blobRow reads a blob's bookkeeping directly, so a test can assert on the
// columns the collector reads rather than on what a helper reports.
func blobRow(t *testing.T, db *DB, d cas.Digest) (refcount int64, present bool, lastRef int64) {
	t.Helper()
	err := db.Reader().QueryRow(
		`SELECT refcount, present, last_ref_at FROM blobs WHERE digest = ?`, d.Bytes()).
		Scan(&refcount, &present, &lastRef)
	if err != nil {
		t.Fatalf("blob %s: %v", d, err)
	}
	return
}

func blobExists(t *testing.T, db *DB, d cas.Digest) bool {
	t.Helper()
	var n int
	if err := db.Reader().QueryRow(
		`SELECT count(*) FROM blobs WHERE digest = ?`, d.Bytes()).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n == 1
}

func TestAllDeploymentRefs(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	a := testProject(t, db, "a")
	b := testProject(t, db, "b")

	want := map[string]int64{}
	for range 3 {
		dep := testDeployment(t, db, a.ID)
		want[dep.PublicID] = a.ID
	}
	dep := testDeployment(t, db, b.ID)
	want[dep.PublicID] = b.ID

	refs, err := db.AllDeploymentRefs(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if len(refs) != len(want) {
		t.Fatalf("got %d refs, want %d", len(refs), len(want))
	}
	for _, ref := range refs {
		pid, ok := want[ref.PublicID]
		if !ok {
			t.Errorf("unexpected public id %q", ref.PublicID)
			continue
		}
		if ref.ProjectID != pid {
			t.Errorf("%s belongs to project %d, want %d", ref.PublicID, ref.ProjectID, pid)
		}
		delete(want, ref.PublicID)
	}
}

func TestDeploymentsInStateSpansProjects(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	a := testProject(t, db, "a")
	b := testProject(t, db, "b")

	// Recovery has to find interrupted deletions wherever they are, so this
	// query is deliberately not project-scoped.
	first := testDeployment(t, db, a.ID)
	second := testDeployment(t, db, b.ID)
	for _, dep := range []*Deployment{first, second} {
		if _, err := db.w.ExecContext(ctx,
			`UPDATE deployments SET state = ? WHERE id = ?`, StateDeleting, dep.ID); err != nil {
			t.Fatal(err)
		}
	}
	testDeployment(t, db, a.ID) // still pending; must not be listed

	got, err := db.DeploymentsInState(ctx, StateDeleting, 0)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 2 {
		t.Fatalf("got %d deleting deployments, want 2", len(got))
	}
	if got[0].ID != first.ID || got[1].ID != second.ID {
		t.Errorf("order is %d,%d; want oldest first (%d,%d)",
			got[0].ID, got[1].ID, first.ID, second.ID)
	}
}

// The active deployment is excluded in SQL rather than filtered by the caller,
// so no retention arithmetic can select the one being served.
func TestInactiveDeploymentsExcludesTheActiveOne(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")
	other := testProject(t, db, "other")
	testDeployment(t, db, other.ID)

	var deps []*Deployment
	for i := range 3 {
		deps = append(deps, ready(t, db, p.ID, string(rune('a'+i))))
	}
	if err := db.ActivateDeployment(ctx, p.ID, deps[1].ID); err != nil {
		t.Fatal(err)
	}

	got, err := db.InactiveDeployments(ctx, p.ID)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 2 {
		t.Fatalf("got %d inactive deployments, want 2", len(got))
	}
	if got[0].ID != deps[2].ID || got[1].ID != deps[0].ID {
		t.Errorf("order is %d,%d; want newest first (%d,%d)",
			got[0].ID, got[1].ID, deps[2].ID, deps[0].ID)
	}
}

func TestExpireStaleDeployments(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	shared := file("shared.js", "shared")
	only := file("only.html", "abandoned")

	// One upload that stalled, one that is merely young, one that finished.
	stale := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, stale.ID, []FileRow{shared, only}); err != nil {
		t.Fatal(err)
	}
	young := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, young.ID, []FileRow{shared}); err != nil {
		t.Fatal(err)
	}
	done := ready(t, db, p.ID, "finished")

	// Age the stalled one past the cutoff. Backdating the row is the only way
	// to test this without the test sleeping.
	if _, err := db.w.ExecContext(ctx,
		`UPDATE deployments SET created_at = ? WHERE id = ?`,
		time.Now().Add(-48*time.Hour).Unix(), stale.ID); err != nil {
		t.Fatal(err)
	}

	n, err := db.ExpireStaleDeployments(ctx, time.Now().Add(-24*time.Hour), "abandoned")
	if err != nil {
		t.Fatal(err)
	}
	if n != 1 {
		t.Fatalf("expired %d deployments, want 1", n)
	}

	got, err := db.DeploymentByPublicID(ctx, p.ID, stale.PublicID)
	if err != nil {
		t.Fatal(err)
	}
	if got.State != StateFailed {
		t.Errorf("state = %q, want %q", got.State, StateFailed)
	}
	if got.Error != "abandoned" {
		t.Errorf("error = %q, want the reason to be recorded", got.Error)
	}

	// Dropping the manifest is the point: it is what takes the refcounts back
	// down so the collector can reach the content.
	if rc, _, _ := blobRow(t, db, only.Digest); rc != 0 {
		t.Errorf("refcount of the abandoned file = %d, want 0", rc)
	}
	// Content the surviving upload also names keeps its reference.
	if rc, _, _ := blobRow(t, db, shared.Digest); rc != 1 {
		t.Errorf("refcount of the shared file = %d, want 1 (the young upload still names it)", rc)
	}

	for _, dep := range []*Deployment{young, done} {
		got, err := db.DeploymentByPublicID(ctx, p.ID, dep.PublicID)
		if err != nil {
			t.Fatal(err)
		}
		if got.State == StateFailed {
			t.Errorf("deployment %s was expired but should not have been", dep.PublicID)
		}
	}

	// Running it again finds nothing left to do.
	n, err = db.ExpireStaleDeployments(ctx, time.Now().Add(-24*time.Hour), "abandoned")
	if err != nil {
		t.Fatal(err)
	}
	if n != 0 {
		t.Errorf("a second pass expired %d more, want 0", n)
	}
}

func TestMarkDeploymentDeleting(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")
	active := ready(t, db, p.ID, "v1")
	spare := ready(t, db, p.ID, "v2")
	if err := db.ActivateDeployment(ctx, p.ID, active.ID); err != nil {
		t.Fatal(err)
	}

	// The one being served is refused, and by the same statement that would
	// have claimed it — there is no window between the check and the claim.
	if err := db.MarkDeploymentDeleting(ctx, active.ID); !errors.Is(err, ErrConflict) {
		t.Fatalf("claiming the active deployment = %v, want ErrConflict", err)
	}
	got, err := db.DeploymentByPublicID(ctx, p.ID, active.PublicID)
	if err != nil {
		t.Fatal(err)
	}
	if got.State != StateReady {
		t.Errorf("the refused claim changed the state to %q", got.State)
	}

	if err := db.MarkDeploymentDeleting(ctx, spare.ID); err != nil {
		t.Fatal(err)
	}
	got, err = db.DeploymentByPublicID(ctx, p.ID, spare.PublicID)
	if err != nil {
		t.Fatal(err)
	}
	if got.State != StateDeleting {
		t.Errorf("state = %q, want %q", got.State, StateDeleting)
	}

	// Re-claiming is fine: a collector that was interrupted after the claim and
	// before the removal has to be able to pick the row up again.
	if err := db.MarkDeploymentDeleting(ctx, spare.ID); err != nil {
		t.Errorf("re-claiming a deployment already being deleted: %v", err)
	}

	if err := db.MarkDeploymentDeleting(ctx, 99999); !errors.Is(err, ErrNotFound) {
		t.Errorf("claiming a missing deployment = %v, want ErrNotFound", err)
	}
}

func TestDeleteDeploymentDropsManifestRefsExplicitly(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	shared := file("shared.js", "shared")
	only := file("index.html", "gone")
	dep := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, dep.ID, []FileRow{shared, only}); err != nil {
		t.Fatal(err)
	}
	keeper := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, keeper.ID, []FileRow{shared}); err != nil {
		t.Fatal(err)
	}

	if err := db.DeleteDeployment(ctx, dep.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := db.DeploymentByPublicID(ctx, p.ID, dep.PublicID); !errors.Is(err, ErrNotFound) {
		t.Errorf("the row survived deletion: %v", err)
	}
	// The AFTER DELETE trigger has to have fired. A cascaded delete would not
	// have fired it, which is why the manifest rows are deleted by hand first.
	if rc, _, _ := blobRow(t, db, only.Digest); rc != 0 {
		t.Errorf("refcount = %d after the only reference was deleted, want 0", rc)
	}
	if rc, _, _ := blobRow(t, db, shared.Digest); rc != 1 {
		t.Errorf("refcount of shared content = %d, want 1", rc)
	}

	if err := db.DeleteDeployment(ctx, dep.ID); !errors.Is(err, ErrNotFound) {
		t.Errorf("deleting again = %v, want ErrNotFound", err)
	}
}

func TestDeleteDeploymentRefusesTheActiveOne(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")
	dep := ready(t, db, p.ID, "v1")
	if err := db.ActivateDeployment(ctx, p.ID, dep.ID); err != nil {
		t.Fatal(err)
	}

	if err := db.DeleteDeployment(ctx, dep.ID); !errors.Is(err, ErrConflict) {
		t.Fatalf("deleting the active deployment = %v, want ErrConflict", err)
	}
	if _, err := db.ActiveDeployment(ctx, p.ID); err != nil {
		t.Errorf("the project stopped serving anything: %v", err)
	}
}

func TestEachPresentBlobAndMarkBlobsAbsent(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	here := file("here.txt", "here")
	gone := file("gone.txt", "gone")
	pending := file("pending.txt", "pending")
	dep := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, dep.ID, []FileRow{here, gone, pending}); err != nil {
		t.Fatal(err)
	}
	for _, f := range []FileRow{here, gone} {
		if err := db.MarkBlobPresent(ctx, f.Digest, f.Size); err != nil {
			t.Fatal(err)
		}
	}

	seen := map[string]bool{}
	if err := db.EachPresentBlob(ctx, func(d cas.Digest) error {
		seen[d.String()] = true
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if len(seen) != 2 || !seen[here.Digest.String()] || !seen[gone.Digest.String()] {
		t.Fatalf("present blobs = %v, want exactly the two uploaded ones", seen)
	}
	if seen[pending.Digest.String()] {
		t.Error("a blob that was never uploaded was reported as present")
	}

	if err := db.MarkBlobsAbsent(ctx, []cas.Digest{gone.Digest}); err != nil {
		t.Fatal(err)
	}
	// The row stays and keeps its reference: the fix for missing content is to
	// ask for it again, not to forget the deployment needs it.
	rc, present, _ := blobRow(t, db, gone.Digest)
	if present {
		t.Error("the blob is still marked present")
	}
	if rc != 1 {
		t.Errorf("refcount = %d, want the manifest reference to survive", rc)
	}
	if _, present, _ := blobRow(t, db, here.Digest); !present {
		t.Error("an unrelated blob was marked absent")
	}

	// An empty list is a no-op rather than a statement with no arguments.
	if err := db.MarkBlobsAbsent(ctx, nil); err != nil {
		t.Errorf("MarkBlobsAbsent(nil): %v", err)
	}

	// The callback's error stops the walk and reaches the caller.
	stop := errors.New("stop")
	if err := db.EachPresentBlob(ctx, func(cas.Digest) error { return stop }); !errors.Is(err, stop) {
		t.Errorf("EachPresentBlob swallowed the callback error: %v", err)
	}
}

func TestUnreferencedBlobs(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	kept := file("kept.txt", "kept")
	dropped := file("dropped.txt", "dropped")
	dep := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, dep.ID, []FileRow{kept, dropped}); err != nil {
		t.Fatal(err)
	}
	if err := db.MarkBlobPresent(ctx, dropped.Digest, dropped.Size); err != nil {
		t.Fatal(err)
	}

	// Referenced content is never listed, however old.
	got, err := db.UnreferencedBlobs(ctx, time.Now().Add(time.Hour), 0)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 0 {
		t.Fatalf("listed %d referenced blobs, want 0", len(got))
	}

	if _, _, err := db.SetManifest(ctx, dep.ID, []FileRow{kept}); err != nil {
		t.Fatal(err)
	}

	// Freshly unreferenced content is protected by the cutoff, which is what
	// gives a request that has already resolved the digest time to open it.
	got, err = db.UnreferencedBlobs(ctx, time.Now().Add(-time.Hour), 0)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 0 {
		t.Fatalf("listed %d blobs inside the grace period, want 0", len(got))
	}

	got, err = db.UnreferencedBlobs(ctx, time.Now().Add(time.Hour), 0)
	if err != nil {
		t.Fatal(err)
	}
	if len(got) != 1 {
		t.Fatalf("listed %d blobs past the cutoff, want 1", len(got))
	}
	if got[0].Digest != dropped.Digest {
		t.Errorf("listed %s, want the dereferenced %s", got[0].Digest, dropped.Digest)
	}
	if got[0].Size != dropped.Size || !got[0].Present {
		t.Errorf("listed blob = %+v, want the size and presence the collector needs", got[0])
	}
}

func TestDeleteBlobRemovesRowAndContentTogether(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	orphan := file("orphan.txt", "orphan")
	held := file("held.txt", "held")
	dep := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, dep.ID, []FileRow{orphan, held}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := db.SetManifest(ctx, dep.ID, []FileRow{held}); err != nil {
		t.Fatal(err)
	}

	removed := 0
	deleted, err := db.DeleteBlob(ctx, orphan.Digest, func() error { removed++; return nil })
	if err != nil {
		t.Fatal(err)
	}
	if !deleted || removed != 1 {
		t.Fatalf("deleted = %v, remove called %d times; want true and once", deleted, removed)
	}
	if blobExists(t, db, orphan.Digest) {
		t.Error("the row survived")
	}

	// A blob that gained a reference since it was listed is left alone, and the
	// content is not touched — this is the check that keeps a deploy racing the
	// collector from losing its files.
	removed = 0
	deleted, err = db.DeleteBlob(ctx, held.Digest, func() error { removed++; return nil })
	if err != nil {
		t.Fatal(err)
	}
	if deleted || removed != 0 {
		t.Errorf("a referenced blob was collected: deleted = %v, remove called %d times", deleted, removed)
	}
	if !blobExists(t, db, held.Digest) {
		t.Error("a referenced blob's row was deleted")
	}

	// Nothing to delete is not an error; the previous sweep already did it.
	deleted, err = db.DeleteBlob(ctx, orphan.Digest, func() error {
		t.Error("remove was called for a row that no longer exists")
		return nil
	})
	if err != nil || deleted {
		t.Errorf("re-deleting = (%v, %v), want (false, nil)", deleted, err)
	}
}

// If the content cannot be removed the row has to survive with it, so the next
// sweep finds the pair again rather than leaving a file nothing points at.
func TestDeleteBlobKeepsTheRowWhenRemovalFails(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	f := file("orphan.txt", "orphan")
	dep := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, dep.ID, []FileRow{f}); err != nil {
		t.Fatal(err)
	}
	if _, _, err := db.SetManifest(ctx, dep.ID, nil); err != nil {
		t.Fatal(err)
	}

	boom := errors.New("disk is having a day")
	deleted, err := db.DeleteBlob(ctx, f.Digest, func() error { return boom })
	if !errors.Is(err, boom) {
		t.Fatalf("DeleteBlob = %v, want the removal error", err)
	}
	if deleted {
		t.Error("reported a deletion that was rolled back")
	}
	if !blobExists(t, db, f.Digest) {
		t.Fatal("the row was deleted even though the content was not")
	}
}

func TestFsckFindsAndRepairsDrift(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	shared := file("shared.js", "shared")
	one := file("one.html", "one")
	two := file("two.html", "two")
	first := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, first.ID, []FileRow{shared, one}); err != nil {
		t.Fatal(err)
	}
	second := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, second.ID, []FileRow{shared, two}); err != nil {
		t.Fatal(err)
	}

	// Triggers maintain these counters, so a healthy store never drifts.
	rep, err := db.Fsck(ctx, false)
	if err != nil {
		t.Fatal(err)
	}
	if rep.Blobs != 3 {
		t.Errorf("examined %d blobs, want 3", rep.Blobs)
	}
	if rep.DriftCount != 0 {
		t.Fatalf("a healthy store reported %d drifting blobs: %+v", rep.DriftCount, rep.Drift)
	}

	// Injecting drift in both directions. Low is the dangerous one: the
	// collector trusts the counter, so a count that reads low is content it
	// will delete while a deployment still names it.
	if _, err := db.w.ExecContext(ctx,
		`UPDATE blobs SET refcount = 0 WHERE digest = ?`, shared.Digest.Bytes()); err != nil {
		t.Fatal(err)
	}
	if _, err := db.w.ExecContext(ctx,
		`UPDATE blobs SET refcount = 7 WHERE digest = ?`, one.Digest.Bytes()); err != nil {
		t.Fatal(err)
	}

	rep, err = db.Fsck(ctx, false)
	if err != nil {
		t.Fatal(err)
	}
	if rep.DriftCount != 2 {
		t.Fatalf("found %d drifting blobs, want 2: %+v", rep.DriftCount, rep.Drift)
	}
	if rep.Repaired != 0 {
		t.Errorf("a report-only check repaired %d rows", rep.Repaired)
	}
	found := map[string]Drift{}
	for _, dr := range rep.Drift {
		found[dr.Digest.String()] = dr
	}
	if dr := found[shared.Digest.String()]; dr.Stored != 0 || dr.Actual != 2 {
		t.Errorf("shared drift = %+v, want stored 0 and actual 2", dr)
	}
	if dr := found[one.Digest.String()]; dr.Stored != 7 || dr.Actual != 1 {
		t.Errorf("one.html drift = %+v, want stored 7 and actual 1", dr)
	}
	// Report-only means exactly that.
	if rc, _, _ := blobRow(t, db, shared.Digest); rc != 0 {
		t.Errorf("refcount = %d, want the injected value left alone", rc)
	}

	rep, err = db.Fsck(ctx, true)
	if err != nil {
		t.Fatal(err)
	}
	if rep.Repaired != 2 {
		t.Errorf("repaired %d rows, want 2", rep.Repaired)
	}
	if rc, _, _ := blobRow(t, db, shared.Digest); rc != 2 {
		t.Errorf("refcount after repair = %d, want the recounted 2", rc)
	}
	if rc, _, _ := blobRow(t, db, one.Digest); rc != 1 {
		t.Errorf("refcount after repair = %d, want the recounted 1", rc)
	}

	rep, err = db.Fsck(ctx, true)
	if err != nil {
		t.Fatal(err)
	}
	if rep.DriftCount != 0 || rep.Repaired != 0 {
		t.Errorf("a repaired store still reports %d drift, %d repaired", rep.DriftCount, rep.Repaired)
	}
}

// The list is for a human to read; the repair is not bounded by it.
func TestFsckCapsTheReportedDriftButRepairsEverything(t *testing.T) {
	ctx := context.Background()
	db := testDB(t)
	p := testProject(t, db, "demo")

	const n = maxReportedDrift + 10
	rows := make([]FileRow, 0, n)
	for i := range n {
		rows = append(rows, file("f"+strconv.Itoa(i)+".txt", "content-"+strconv.Itoa(i)))
	}
	dep := testDeployment(t, db, p.ID)
	if _, _, err := db.SetManifest(ctx, dep.ID, rows); err != nil {
		t.Fatal(err)
	}
	if _, err := db.w.ExecContext(ctx, `UPDATE blobs SET refcount = 42`); err != nil {
		t.Fatal(err)
	}

	rep, err := db.Fsck(ctx, true)
	if err != nil {
		t.Fatal(err)
	}
	if rep.DriftCount != n {
		t.Errorf("DriftCount = %d, want the full %d", rep.DriftCount, n)
	}
	if len(rep.Drift) != maxReportedDrift {
		t.Errorf("listed %d drifting blobs, want the report capped at %d", len(rep.Drift), maxReportedDrift)
	}
	if rep.Repaired != n {
		t.Errorf("repaired %d rows, want all %d", rep.Repaired, n)
	}
	if rep, err = db.Fsck(ctx, false); err != nil || rep.DriftCount != 0 {
		t.Errorf("after repair: %d drift, err %v", rep.DriftCount, err)
	}
}