mesh-catalog/modules/distribution/cmd/distribution/main.go

// Command distribution is distribution's program (novox/hq design 51): it serves the artifact store's port in front of
// the registry container, which listens on its node's loopback only. Reads (GET, HEAD) go through to every node's
// container runtime; a write is refused but while the direct-write window is open, never over a tag the seat holds
// published, or with the bootstrap's one-off credential (holder.Front). The windows are distribution's state windows:
// the program moves them (windows.go) and follows the direct-write window's record on each change; it holds no timer.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"log"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"strings"
	"sync/atomic"
	"syscall"
	"time"

	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats.go/jetstream"

	"distribution/internal/holder"
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	if err := run(ctx, os.Getenv); err != nil {
		fmt.Fprintln(os.Stderr, "distribution:", err)
		os.Exit(1)
	}
}

func run(ctx context.Context, getenv func(string) string) error {
	logger := log.New(os.Stderr, "distribution: ", log.LstdFlags)
	store, err := url.Parse(or(getenv("MESH_REGISTRY_LOOPBACK"), "http://127.0.0.1:5001"))
	if err != nil {
		return err
	}
	oci := &holder.OCI{Base: store}
	var open atomic.Bool
	open.Store(true) // until the window's state says it closed: no wider than today's store
	cred, err := process.ReadBrokerAccount(getenv)
	if err != nil {
		return err
	}
	conn, js, err := cred.Dial()
	if err != nil {
		return err
	}
	defer conn.Close()
	// The direct-write window's record, written by the tools bundle on distribution's own account.
	windows := process.OptionalBucket(ctx, js, "distribution_windows", "the direct-write window is read as open until the program starts again", logger.Printf)
	if windows != nil {
		go follow(ctx, windows, &open, logger)
		go followBootstrap(ctx, windows, logger)
		calls, err := process.Connect(cred)
		if err != nil {
			return err
		}
		defer calls.Close()
		callers, err := holder.ParseCallers(getenv("MESH_CONFIG_CALLERS"))
		if err != nil {
			return err
		}
		findings := process.OptionalBucket(ctx, js, "distribution_findings", "a window held for seven days is said in the log only", logger.Printf)
		go drive(ctx, js, cred, &holder.Driver{Delivery: busDelivery{calls}, State: kvWindows{windows}, Callers: callers,
			Closed: sayClosed(js, cred), Finding: findingsKV(findings, logger.Printf)}, logger)
	}
	front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf,
		Published: func(name, tag string) (bool, error) {
			d, err := oci.Digest(name, tag)
			return d != "", err
		},
		Bootstrap: func() (string, bool) {
			b, err := os.ReadFile(holder.BootstrapFile)
			if err != nil {
				return "", false
			}
			return strings.TrimSpace(string(b)), true
		}}
	srv := &http.Server{Addr: or(getenv("MESH_REGISTRY_LISTEN"), ":5000"), Handler: front, ReadHeaderTimeout: 30 * time.Second}
	errs := make(chan error, 1)
	go func() { errs <- srv.ListenAndServe() }()
	logger.Printf("serving the store's port %s in front of %s", srv.Addr, store)
	select {
	case <-ctx.Done():
		shut, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
		defer cancel()
		return srv.Shutdown(shut)
	case err := <-errs:
		if errors.Is(err, http.ErrServerClosed) {
			return nil
		}
		return err
	}
}

// follow keeps open as the direct-write window's record says, on every change of it (a watch, no timer).
func follow(ctx context.Context, kv jetstream.KeyValue, open *atomic.Bool, logger *log.Logger) {
	w, err := kv.Watch(ctx, holder.DirectWritesWindow)
	if err != nil {
		logger.Printf("the direct-write window cannot be watched (%v): read as open", err)
		return
	}
	for e := range w.Updates() {
		if e == nil || e.Operation() != jetstream.KeyValuePut {
			continue
		}
		var win holder.Window
		if json.Unmarshal(e.Value(), &win) != nil {
			continue
		}
		closed := !win.Open && !win.Closed.IsZero()
		if closed && open.Swap(false) {
			logger.Printf("the direct-write window is closed: every write but distribution's own is refused, naming publish")
		}
	}
}

// followBootstrap removes the bootstrap's one-off credential (holder.BootstrapFile) when the tools bundle closes the
// window bootstrap-push at its first answered publish (design 51 *On a fresh mesh*): the program runs as distribution's
// account, which owns the file. A file already gone is no fault; one that cannot be removed is said each time the
// window is written.
func followBootstrap(ctx context.Context, kv jetstream.KeyValue, logger *log.Logger) {
	w, err := kv.Watch(ctx, holder.BootstrapWindow)
	if err != nil {
		logger.Printf("the bootstrap's credential window cannot be watched (%v): its file stays until the program starts again", err)
		return
	}
	for e := range w.Updates() {
		if e == nil || e.Operation() != jetstream.KeyValuePut {
			continue
		}
		var win holder.Window
		if json.Unmarshal(e.Value(), &win) != nil || win.Closed.IsZero() {
			continue
		}
		if why := removeBootstrap(holder.BootstrapFile); why != "" {
			logger.Printf("%s", why)
		}
	}
}

// removeBootstrap removes the credential's file; it answers what went wrong, or "" when the file is gone.
func removeBootstrap(path string) string {
	switch err := os.Remove(path); {
	case err == nil:
		return "the bootstrap's one-off credential is discarded: the holders serve publish"
	case errors.Is(err, fs.ErrNotExist):
		return ""
	default:
		return fmt.Sprintf("the bootstrap's one-off credential could not be discarded, so it is still accepted: %v", err)
	}
}

func or(v, d string) string {
	if strings.TrimSpace(v) == "" {
		return d
	}
	return v
}