mesh-controller/cmd/mesh-controller/collect.go

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"time"

	"github.com/novox/mesh-controller/internal/artifacts"
	"github.com/novox/mesh-controller/internal/inventory"
)

// Letting the artifact store go of what the mesh no longer keeps (novox/hq ADR 0189, issue 108).
//
// **Run where the records change.** A build is the moment new bytes landed in the store and the
// moment the keep set moved, so it is the moment to say what may go — and it needs no timer of
// its own. Reclaiming the bytes is the store's own nightly step; this only decides.
//
// Never fatal to a build. The build succeeded, the module is registered, and a store that could
// not be reached is a thing to say rather than a reason to undo any of that. The next build asks
// again, and the references it could not collect are still uncollected, so nothing is lost by
// having failed.

// collect asks the store to let go of everything the mesh made and no longer keeps, and records
// what it let go of. Says what it did and what it could not; returns nothing, because nothing
// upstream should branch on it.
func collect(ctx context.Context, inv *inventory.Inventory) {
	references, err := inv.ToCollect(ctx)
	if err != nil {
		fmt.Fprintf(os.Stderr, "could not work out what the artifact store may let go of: %v\n", err)
		return
	}
	kept, err := inv.KeptArchives(ctx)
	if err != nil {
		fmt.Fprintf(os.Stderr, "could not work out which archives the artifact store keeps: %v\n", err)
		return
	}
	if len(references) == 0 && len(kept) == 0 {
		return
	}
	shelf, err := inv.Catalogue(ctx)
	if err != nil {
		fmt.Fprintf(os.Stderr, "could not read the catalogue to find the artifact store: %v\n", err)
		return
	}
	// As the mesh reaches it from the network. Empty means the store is not on the network — on a
	// mesh being raised it is not yet, and there the store holds one build of anything and has
	// nothing to collect.
	address, err := artifactStoreAddress(ctx, inv, shelf, "")
	if err != nil || address == "" {
		if err != nil {
			fmt.Fprintf(os.Stderr, "could not find the artifact store to collect from: %v\n", err)
		}
		return
	}

	// **Bounded, because this runs inside somebody's build.** The first sweep of a mesh that has
	// never collected has the whole history to get through, and a person waiting on `build` should
	// not pay for it. Two bounds, and what is left over is simply offered again next time —
	// builds are frequent, and the point is that the store stops growing, not that it empties
	// tonight.
	within, stop := context.WithTimeout(ctx, sweepBudget)
	defer stop()
	store := artifacts.Store{Address: address}

	// **Hold before letting go** (novox/hq issue 253). The store's collector keeps only what a
	// manifest names, and archives were published as bare blobs, so every kept archive is first
	// held by its manifest — which backfills the ones published before holders, a few at a time
	// as builds come, and is two HEADs each once done. A kept archive that could not be held stops
	// the sweep before it deletes anything: "everything kept is held" is the precondition the
	// collector's safety rests on, and a store refusing a hold would refuse the deletes too.
	wrote, missing, err := holdKept(within, store, kept)
	if wrote > 0 {
		fmt.Fprintf(os.Stderr, "the artifact store now holds %d more kept archive(s) by a manifest\n", wrote)
	}
	if missing > 0 {
		fmt.Fprintf(os.Stderr, "%d archive(s) the mesh keeps are not in the artifact store at all; "+
			"`collection` lists them\n", missing)
	}
	if err != nil {
		fmt.Fprintf(os.Stderr, "not every kept archive could be held, so nothing was let go: %v\n", err)
		return
	}

	var done []string
	var left, skipped int
	for i, reference := range references {
		if i >= mostPerSweep || within.Err() != nil {
			left = len(references) - i
			break
		}
		err := store.LetGo(within, reference)
		if err == nil || errors.Is(err, artifacts.Gone) {
			// Gone is the outcome wanted, already true. Recorded so the next sweep does not ask
			// again for ever.
			done = append(done, reference)
			continue
		}
		if errors.Is(err, artifacts.ErrNotOurs) {
			// **A fact about this record, so this record is skipped** (novox/hq issue 226). Not
			// marked collected — the mesh did not remove it and should not claim to — and not a
			// reason to stop, because the store was never asked. One of these at the front of
			// the oldest-first order ended every sweep until this.
			skipped++
			if skipped == 1 {
				fmt.Fprintf(os.Stderr,
					"the sweep will not address %s and went on: %v\n", reference, err)
			}
			continue
		}
		// **Stopped at the first refusal by the STORE, not pushed through.** A store that refuses
		// one refuses all of them — deletion disabled, the store down, the network gone — so
		// going on would be a hundred identical failures and a hundred identical log lines in
		// front of whoever was building something.
		fmt.Fprintf(os.Stderr, "the artifact store kept %s, so nothing more was asked of it: %v\n",
			reference, err)
		left = len(references) - i
		break
	}

	if len(done) > 0 {
		// Recorded outside `within`: the deletions happened, and losing the record of them because
		// the sweep ran out of budget would mean asking about them again for ever.
		if err := inv.MarkCollected(ctx, done); err != nil {
			fmt.Fprintf(os.Stderr, "the store let go of %d artifact(s) and the record of it did not keep: %v\n",
				len(done), err)
			return
		}
		fmt.Fprintf(os.Stderr, "the artifact store let go of %d artifact(s) the mesh no longer keeps\n",
			len(done))
	}
	if left > 0 {
		fmt.Fprintf(os.Stderr, "%d more to collect; the next build asks again\n", left)
	}
	if skipped > 0 {
		fmt.Fprintf(os.Stderr, "%d artifact(s) the sweep will not address were skipped\n", skipped)
	}
}

// holdKept holds every kept archive by its manifest, stopping at the first refusal by the store.
// Answers how many holders it wrote and how many kept archives the store does not have.
//
// A missing archive is counted rather than fatal: there is nothing to hold, and that is a fact
// for an operator to read (`collection`), not a reason to stop collecting what is not kept. A
// reference the store cannot be asked about is skipped as the deletion loop skips one
// (novox/hq issue 226).
func holdKept(ctx context.Context, store artifacts.Store, kept []string) (wrote, missing int, err error) {
	for _, reference := range kept {
		if err := ctx.Err(); err != nil {
			return wrote, missing, fmt.Errorf("ran out of time before %s: %w", reference, err)
		}
		did, err := store.Hold(ctx, reference)
		switch {
		case err == nil:
			if did {
				wrote++
			}
		case errors.Is(err, artifacts.Gone):
			missing++
		case errors.Is(err, artifacts.ErrNotOurs):
		default:
			return wrote, missing, fmt.Errorf("holding %s: %w", reference, err)
		}
	}
	return wrote, missing, nil
}

// mostPerSweep is how many artifacts one sweep will ask about. Enough that a mesh building
// several times a day converges within days of this landing; small enough that no single build
// waits on the whole backlog.
const mostPerSweep = 200

// sweepBudget is the longest a sweep will keep a build waiting.
const sweepBudget = 60 * time.Second