path: root/cmd/mbf/internal/ci/ci.go

// 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 cancelled.
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
		}
	}
}

// archiveTime is the hardcoded time written to every header time field.
var archiveTime = time.Unix(0, 0)

// 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.AccessTime, h.ChangeTime, h.ModTime = archiveTime, archiveTime, archiveTime
		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
}

const (
	// endpointDist accepts a Path.writeSource stream and produces a
	// distribution for the submitted source.
	endpointDist = "/dist"

	// endpointRace is like endpointDist, but the resulting distribution is
	// instrumented with the data race detector.
	endpointRace = "/race"
)

// 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, endpointDist, source, version)
	if err != nil {
		return err
	}
	return p.fetch(ctx, id, output)
}

// MakeDistRace creates a hakurei distribution (race) using the CI service.
func (p Path) MakeDistRace(
	ctx context.Context,
	w io.Writer,
	output, source, version string,
) error {
	id, err := p.writeSource(ctx, w, endpointRace, 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 "+endpointDist, 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()),
			)
		}
	})
	mux.HandleFunc("POST "+endpointRace, 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-race"))
		if err = cure(c, a, w); err != nil {
			msg.Verbose(err)
			return
		}
		if msg.IsVerbose() {
			msg.Verbosef(
				"satisfied distribution %s (race)",
				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
}