From b94e6faa9d20dab9b179ee40b5f749c7f13d3f9a Mon Sep 17 00:00:00 2001 From: Ophestra Date: Tue, 29 Sep 2026 23:56:37 +0900 Subject: cmd/mbf: proof-of-concept CI service 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 --- cmd/mbf/internal/ci/ci.go | 522 +++++++++++++++++++++++++++++++++++++++++ cmd/mbf/internal/ci/ci_test.go | 55 +++++ 2 files changed, 577 insertions(+) create mode 100644 cmd/mbf/internal/ci/ci.go create mode 100644 cmd/mbf/internal/ci/ci_test.go (limited to 'cmd/mbf/internal/ci') 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) + } +} -- cgit v1.3.1