cmd/mbf: proof-of-concept CI servicedevelop

This accepts CI work over a shared read-only cache directory. This
replaces the nix CI service and prepares the removal of nix from the
test infrastructure.

Signed-off-by: Ophestra <cat@gensokyo.uk>

3 files changed, 613 insertions, 1 deletions

diff --git a/cmd/mbf/internal/ci/ci.go b/cmd/mbf/internal/ci/ci.go
new file mode 100644
index 00000000..c9d585d4
--- /dev/null
+++ b/cmd/mbf/internal/ci/ci.go
@@ -0,0 +1,522 @@
+// Package ci implements the CI service and client.
+package ci
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"crypto/ed25519"
+	"encoding/binary"
+	"errors"
+	"io"
+	"io/fs"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"path"
+	"path/filepath"
+	"sync"
+	"syscall"
+	"time"
+	"unique"
+	"unsafe"
+	_ "unsafe" // for go:linkname
+
+	"hakurei.app/internal/rosa"
+	"hakurei.app/message"
+	"hakurei.app/pkg"
+)
+
+// setSource is made available here to accept prepared hakurei tarballs.
+//
+//go:linkname setSource hakurei.app/internal/rosa.(*S).setSource
+func setSource(s *rosa.S, p []byte, version string)
+
+// inotifyInit returns a new inotify instance.
+func inotifyInit() (*os.File, error) {
+	fd, err := syscall.InotifyInit1(syscall.IN_NONBLOCK | syscall.IN_CLOEXEC)
+	if err != nil {
+		return nil, os.NewSyscallError("inotify_init1", err)
+	}
+	return os.NewFile(uintptr(fd), "inotify"), nil
+}
+
+// inotifyAddWatch adds pathname to in.
+func inotifyAddWatch(
+	in *os.File,
+	pathname string,
+	mask uint32,
+) (watchdesc int, err error) {
+	sc, _err := in.SyscallConn()
+	if _err != nil {
+		return -1, _err
+	}
+	if _err = sc.Control(func(fd uintptr) {
+		watchdesc, err = syscall.InotifyAddWatch(int(fd), pathname, mask)
+	}); _err != nil {
+		return -1, _err
+	}
+	return
+}
+
+// Follow reads from the file at pathname and writes its contents and any new
+// contents to w. follow returns if a read, write or inotify error occurs, or
+// the context is canceled.
+func Follow(ctx context.Context, pathname string, w io.Writer) (err error) {
+	var in *os.File
+	if in, err = inotifyInit(); err != nil {
+		return
+	}
+	defer func() {
+		if _err := in.Close(); err == nil {
+			err = _err
+		}
+	}()
+
+	if _, err = inotifyAddWatch(in, pathname, syscall.IN_MODIFY); err != nil {
+		return
+	}
+	var r *os.File
+	if r, err = os.Open(pathname); err != nil {
+		return
+	}
+	defer func() {
+		if _err := r.Close(); err == nil {
+			err = _err
+		}
+	}()
+
+	done := make(chan struct{})
+	defer close(done)
+	go func() {
+		select {
+		case <-ctx.Done():
+			now := time.Now()
+			_ = in.SetDeadline(now)
+			return
+
+		case <-done:
+			return
+		}
+	}()
+
+	if _, err = io.Copy(w, r); err != nil {
+		return
+	}
+
+	buf := make([]byte, os.Getpagesize())
+	for {
+		if _, err = io.Copy(w, r); err != nil {
+			return
+		}
+		if _, err = in.Read(buf); err != nil {
+			if errors.Is(err, os.ErrDeadlineExceeded) {
+				err = nil
+			}
+			return
+		}
+	}
+}
+
+// Path wraps the [pkg.Cache] pathname.
+type Path string
+
+// String returns the value of p.
+func (p Path) String() string { return string(p) }
+
+// append is [filepath.Join] with p as the first element.
+func (p Path) append(elem ...string) string {
+	return filepath.Join(append([]string{p.String()}, elem...)...)
+}
+
+// New returns a new [Path].
+func New(c *pkg.Cache) Path { return Path(c.Path().String()) }
+
+// name returns the CI socket pathname.
+func (p Path) name() string { return p.append("ci") }
+
+// client returns the CI http client.
+func (p Path) client() *http.Client {
+	var d net.Dialer
+	addr := net.UnixAddr{
+		Net:  "unix",
+		Name: p.name(),
+	}
+
+	return &http.Client{Transport: &http.Transport{
+		DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+			return d.DialUnix(ctx, "unix", nil, &addr)
+		},
+	}}
+}
+
+// cure writes the identifier of the pending artifact, cures the artifact, and
+// writes the cure whence. For an unsuccessful cure, a negative whence is
+// written, followed by a user-facing error string.
+func cure(c *pkg.Cache, a pkg.Artifact, w http.ResponseWriter) error {
+	h := w.Header()
+	h.Set("Content-Type", "application/octet-stream")
+	h.Set("Cache-Control", "no-cache")
+
+	f, ok := w.(http.Flusher)
+	if !ok {
+		_, _ = w.Write([]byte{0})
+		return errors.ErrUnsupported
+	}
+
+	id := c.Ident(a).Value()
+	if _, err := w.Write(id[:]); err != nil {
+		return err
+	}
+	f.Flush()
+
+	_, _, whence, err := c.CureWhence(a)
+	if err != nil {
+		whence = -1
+	}
+	if _, _err := w.Write(
+		binary.LittleEndian.AppendUint64(nil, uint64(whence)),
+	); _err != nil {
+		return _err
+	}
+
+	if err != nil {
+		if _, _err := io.WriteString(w, err.Error()); _err != nil {
+			return _err
+		}
+	}
+	f.Flush()
+	return err
+}
+
+// spool holds reusable [rosa.S] instances.
+var spool = sync.Pool{New: func() any { return rosa.New() }}
+
+// getS returns the address of a populated [rosa.S]. Its hakurei-source may be
+// clobbered and must be replaced using setSource before use.
+func getS() *rosa.S { return spool.Get().(*rosa.S) }
+
+// putS returns s to spool.
+func putS(s *rosa.S) { spool.Put(s) }
+
+// versionSize is the maximum size of the specified version string, plus its
+// deliminator byte.
+const versionSize = 8 + 16 + 6 + 2
+
+// errBadVersion is returned by readSource if a header does not contain
+// the deliminator byte.
+var errBadVersion = errors.New("unterminated version string")
+
+// readSource reads a version string and compressed source tarball from r and
+// returns the address of a [rosa.S] with this source tarball. The resulting
+// [rosa.S] must be returned via putS.
+func readSource(w http.ResponseWriter, r *http.Request) (*rosa.S, error) {
+	var header [versionSize]byte
+	_, err := io.ReadFull(r.Body, header[:])
+	if err != nil {
+		_ = r.Body.Close()
+		http.Error(w, "bad header", http.StatusBadRequest)
+		return nil, err
+	}
+
+	var version string
+	if i := bytes.IndexByte(header[:], 0); i < 0 {
+		_ = r.Body.Close()
+		http.Error(w, "unterminated version string", http.StatusBadRequest)
+		return nil, errBadVersion
+	} else {
+		version = unsafe.String(&header[0], i)
+	}
+
+	var p []byte
+	if p, err = io.ReadAll(r.Body); err != nil {
+		_ = r.Body.Close()
+		http.Error(w, "cannot receive payload", http.StatusInternalServerError)
+		return nil, err
+	}
+
+	s := getS()
+	setSource(s, p, version)
+	return s, r.Body.Close()
+}
+
+// ErrDaemonError is returned by writeSource generally if cure could not flush
+// on the connection to notify completion.
+var ErrDaemonError = errors.New("CI service could not process the request")
+
+// writeSource writes a source tarball to the specified endpoint of the CI
+// backend servicing the cache referred to by cm.
+func (p Path) writeSource(
+	ctx context.Context,
+	w io.Writer,
+	endpoint, source, version string,
+) (*pkg.ID, error) {
+	if len(version) >= versionSize {
+		return nil, syscall.ENOMEM
+	}
+	var header [versionSize]byte
+	copy(header[:], version[:])
+
+	var buf bytes.Buffer
+	gw := gzip.NewWriter(&buf)
+	tw := tar.NewWriter(gw)
+
+	if err := filepath.WalkDir(source, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+
+		if d.IsDir() && d.Name() == ".git" {
+			return fs.SkipDir
+		}
+
+		var fi fs.FileInfo
+		if fi, err = d.Info(); err != nil {
+			return err
+		}
+
+		var linkname string
+		if fi.Mode()&fs.ModeSymlink != 0 {
+			if linkname, err = os.Readlink(path); err != nil {
+				return err
+			}
+		}
+
+		var h *tar.Header
+		if h, err = tar.FileInfoHeader(fi, linkname); err != nil {
+			return err
+		}
+		h.Name = path
+
+		var isVersion bool
+		if dir, file := filepath.Split(path); filepath.Base(dir) == "dist" &&
+			file == "VERSION" && fi.Mode().IsRegular() {
+			isVersion = true
+			h.Size = int64(len(version))
+		}
+
+		if err = tw.WriteHeader(h); err != nil {
+			return err
+		}
+
+		if isVersion {
+			_, err = io.WriteString(tw, version)
+			return err
+		}
+
+		if fi.Mode().IsRegular() {
+			var f io.ReadCloser
+			if f, err = os.Open(path); err != nil {
+				return err
+			}
+
+			_, err = io.Copy(tw, f)
+			if _err := f.Close(); err == nil {
+				err = _err
+			}
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	if err := tw.Close(); err != nil {
+		return nil, err
+	}
+	if err := gw.Close(); err != nil {
+		return nil, err
+	}
+
+	req, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodPost,
+		"http://"+path.Join("_", endpoint),
+		io.MultiReader(bytes.NewReader(header[:]), bytes.NewReader(buf.Bytes())),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	var resp *http.Response
+	resp, err = p.client().Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	var id pkg.ID
+	_, err = io.ReadFull(resp.Body, id[:])
+	if err != nil {
+		_ = resp.Body.Close()
+		if errors.Is(err, io.ErrUnexpectedEOF) {
+			return nil, ErrDaemonError
+		}
+		return nil, err
+	}
+
+	c, cancel := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	var whence int
+	go func() {
+		defer cancel()
+		var wbuf [8]byte
+		_, _err := io.ReadFull(resp.Body, wbuf[:])
+		whence = int(binary.LittleEndian.Uint64(wbuf[:]))
+		done <- _err
+	}()
+
+	if w != nil {
+	retry:
+		err = Follow(c, filepath.Join(p.String(), "status", pkg.Encode(id)), w)
+		if err != nil {
+			if ctx.Err() == nil && errors.Is(err, os.ErrNotExist) {
+				goto retry
+			}
+
+			_ = resp.Body.Close()
+			return nil, err
+		}
+	}
+
+	err = <-done
+	if err != nil {
+		_ = resp.Body.Close()
+		return nil, err
+	}
+
+	if whence < 0 {
+		var m []byte
+		if m, err = io.ReadAll(resp.Body); err != nil {
+			_ = resp.Body.Close()
+			return nil, err
+		} else if err = resp.Body.Close(); err != nil {
+			return nil, err
+		}
+		return nil, errors.New(unsafe.String(unsafe.SliceData(m), len(m)))
+	}
+	return &id, resp.Body.Close()
+}
+
+// The stubKey is used by the mirror service exposed by serve where
+// authentication is unnecessary.
+var stubKey = ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize))
+
+// fetch fetches the outcome of id and writes it to the specified directory.
+func (p Path) fetch(ctx context.Context, id *pkg.ID, output string) error {
+	c := p.client()
+	r, err := rosa.NewRemote(
+		c, "http://_",
+		stubKey.Public().(ed25519.PublicKey),
+	)
+	if err != nil {
+		return err
+	}
+
+	var sum *pkg.Checksum
+	if sum, err = r.Artifact(ctx, unique.Make(*id)); err != nil {
+		return err
+	}
+
+	var req *http.Request
+	if req, err = http.NewRequestWithContext(
+		ctx,
+		http.MethodGet,
+		"http://"+path.Join("_", "outcome", pkg.Encode(*sum)),
+		nil,
+	); err != nil {
+		return err
+	}
+
+	var resp *http.Response
+	if resp, err = c.Do(req); err != nil {
+		return err
+	}
+
+	err = pkg.Extract(resp.Body, output, nil)
+	if closeErr := resp.Body.Close(); err == nil {
+		err = closeErr
+	}
+	return err
+}
+
+// MakeDist creates a hakurei distribution using the CI service.
+func (p Path) MakeDist(
+	ctx context.Context,
+	w io.Writer,
+	output, source, version string,
+) error {
+	id, err := p.writeSource(ctx, w, "/dist", source, version)
+	if err != nil {
+		return err
+	}
+	return p.fetch(ctx, id, output)
+}
+
+// Serve services CI workload dispatched to c.
+func Serve(ctx context.Context, msg message.Msg, c *pkg.Cache) error {
+	const shutdownTimeout = 15 * time.Second
+	p := New(c)
+	addr := net.UnixAddr{
+		Net:  "unix",
+		Name: p.name(),
+	}
+
+	var mux http.ServeMux
+	mux.HandleFunc("POST /dist", func(w http.ResponseWriter, r *http.Request) {
+		s, err := readSource(w, r)
+		if err != nil {
+			msg.Verbose(err)
+			return
+		}
+		defer putS(s)
+
+		_, a := s.Std().MustLoad(rosa.H("hakurei-dist"))
+		if err = cure(c, a, w); err != nil {
+			msg.Verbose(err)
+			return
+		}
+		if msg.IsVerbose() {
+			msg.Verbosef(
+				"satisfied distribution %s",
+				pkg.Encode(c.Ident(a).Value()),
+			)
+		}
+	})
+
+	if r, err := os.OpenRoot(p.String()); err != nil {
+		return err
+	} else {
+		defer func() {
+			if err = r.Close(); err != nil {
+				msg.Verbose(err)
+			}
+		}()
+		rosa.NewMirror(msg, r.FS(), stubKey).Register(&mux)
+	}
+
+	server := http.Server{Handler: &mux}
+	go func() {
+		<-ctx.Done()
+		cc, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
+		defer cancel()
+		if _err := server.Shutdown(cc); _err != nil {
+			log.Fatal(_err)
+		}
+	}()
+
+	ul, err := net.ListenUnix("unix", &addr)
+	if err != nil {
+		return err
+	}
+	ul.SetUnlinkOnClose(true)
+	msg.Verbosef("listening on %s", addr.Name)
+
+	err = server.Serve(ul)
+	if errors.Is(err, http.ErrServerClosed) {
+		err = nil
+	}
+	return err
+}
diff --git a/cmd/mbf/internal/ci/ci_test.go b/cmd/mbf/internal/ci/ci_test.go
new file mode 100644
index 00000000..b5ed134e
--- /dev/null
+++ b/cmd/mbf/internal/ci/ci_test.go
@@ -0,0 +1,55 @@
+package ci_test
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"hakurei.app/cmd/mbf/internal/ci"
+)
+
+func TestFollow(t *testing.T) {
+	t.Parallel()
+
+	pathname := filepath.Join(t.TempDir(), "f")
+	w, err := os.Create(pathname)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var buf bytes.Buffer
+	ctx, cancel := context.WithCancel(t.Context())
+
+	var want string
+	go func() {
+		defer cancel()
+		for _, s := range []string{
+			"\xde\xad\xbe\xef",
+			"\xff\xff\xff\xff",
+			"\x00\x00",
+		} {
+			want += s
+			if _, _err := w.WriteString(s); _err != nil {
+				panic(_err)
+			}
+		}
+	}()
+
+retry:
+	if err = ci.Follow(ctx, pathname, &buf); err != nil {
+		t.Fatal(err)
+	}
+	<-ctx.Done()
+
+	// the inotify event takes time to arrive, and there is no way to
+	// synchronise for this cleanly
+	if buf.Len() != len(want) {
+		goto retry
+	}
+
+	if got := buf.String(); got != want {
+		t.Fatalf("follow: %q, want %q", got, want)
+	}
+}
diff --git a/cmd/mbf/main.go b/cmd/mbf/main.go
index 22bfbc3e..014a7f47 100644
--- a/cmd/mbf/main.go
+++ b/cmd/mbf/main.go
@@ -43,6 +43,7 @@ import (
 	"hakurei.app/message"
 	"hakurei.app/pkg"
 
+	"hakurei.app/cmd/mbf/internal/ci"
 	"hakurei.app/cmd/mbf/internal/pkgsite"
 	"hakurei.app/cmd/mbf/internal/pkgsite/ui"
 )
@@ -504,7 +505,7 @@ func main() {
 	c.NewCommand(
 		"daemon",
 		"Service artifact IR with Rosa OS extensions",
-		func(args []string) error {
+		func([]string) error {
 			ul, err := net.ListenUnix("unix", &addr)
 			if err != nil {
 				return err
@@ -514,6 +515,40 @@ func main() {
 		},
 	)
 
+	_ci := c.New("ci", command.UsageInternal)
+	_ci.NewCommand(
+		"daemon",
+		"Service CI workload dispatched through the socket",
+		func([]string) error {
+			return cm.Do(func(cache *pkg.Cache) error {
+				return ci.Serve(ctx, msg, cache)
+			})
+		},
+	)
+	{
+		var flagOutput string
+		_ci.NewCommand(
+			"dist",
+			"Request distribution tarball for the specified source directory",
+			func(args []string) error {
+				if len(args) != 2 {
+					return errors.New("dist requires 2 arguments")
+				}
+
+				return ci.Path(cm.base).MakeDist(
+					ctx,
+					os.Stdout,
+					flagOutput,
+					args[0], args[1],
+				)
+			},
+		).Flag(
+			&flagOutput,
+			"o", command.StringFlag("."),
+			"Write the resulting distribution to the named directory",
+		)
+	}
+
 	c.NewCommand(
 		"keygen",
 		"Create keypair for local cache",