about summary refs log tree commit diff
path: root/tvix/nar-bridge/pkg/server
diff options
context:
space:
mode:
Diffstat (limited to 'tvix/nar-bridge/pkg/server')
-rw-r--r--tvix/nar-bridge/pkg/server/blob_upload.go48
-rw-r--r--tvix/nar-bridge/pkg/server/directory_upload.go66
-rw-r--r--tvix/nar-bridge/pkg/server/nar_get.go212
-rw-r--r--tvix/nar-bridge/pkg/server/nar_put.go140
-rw-r--r--tvix/nar-bridge/pkg/server/narinfo_get.go146
-rw-r--r--tvix/nar-bridge/pkg/server/narinfo_put.go174
-rw-r--r--tvix/nar-bridge/pkg/server/server.go86
-rw-r--r--tvix/nar-bridge/pkg/server/util.go24
8 files changed, 896 insertions, 0 deletions
diff --git a/tvix/nar-bridge/pkg/server/blob_upload.go b/tvix/nar-bridge/pkg/server/blob_upload.go
new file mode 100644
index 000000000000..fe554f5a5a2b
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/blob_upload.go
@@ -0,0 +1,48 @@
+package server
+
+import (
+	storev1pb "code.tvl.fyi/tvix/store/protos"
+	"context"
+	"encoding/base64"
+	"fmt"
+	log "github.com/sirupsen/logrus"
+	"io"
+)
+
+// this returns a callback function that can be used as fileCb
+// for the reader.Import function call
+func genBlobServiceWriteCb(ctx context.Context, blobServiceClient storev1pb.BlobServiceClient) func(io.Reader) error {
+	return func(fileReader io.Reader) error {
+		// Read from fileReader into a buffer.
+		// We currently buffer all contents and send them to blobServiceClient at once,
+		// but that's about to change.
+		contents, err := io.ReadAll(fileReader)
+		if err != nil {
+			return fmt.Errorf("unable to read all contents from file reader: %w", err)
+		}
+
+		log := log.WithField("blob_size", len(contents))
+
+		log.Infof("about to upload blob")
+
+		putter, err := blobServiceClient.Put(ctx)
+		if err != nil {
+			// return error to the importer
+			return fmt.Errorf("error from blob service: %w", err)
+		}
+		err = putter.Send(&storev1pb.BlobChunk{
+			Data: contents,
+		})
+		if err != nil {
+			return fmt.Errorf("putting blob chunk: %w", err)
+		}
+		resp, err := putter.CloseAndRecv()
+		if err != nil {
+			return fmt.Errorf("close blob putter: %w", err)
+		}
+
+		log.WithField("digest", base64.StdEncoding.EncodeToString(resp.GetDigest())).Info("uploaded blob")
+
+		return nil
+	}
+}
diff --git a/tvix/nar-bridge/pkg/server/directory_upload.go b/tvix/nar-bridge/pkg/server/directory_upload.go
new file mode 100644
index 000000000000..02b173698042
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/directory_upload.go
@@ -0,0 +1,66 @@
+package server
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+
+	storev1pb "code.tvl.fyi/tvix/store/protos"
+	log "github.com/sirupsen/logrus"
+)
+
+type DirectoriesUploader struct {
+	ctx                       context.Context
+	directoryServiceClient    storev1pb.DirectoryServiceClient
+	directoryServicePutStream storev1pb.DirectoryService_PutClient
+}
+
+func NewDirectoriesUploader(ctx context.Context, directoryServiceClient storev1pb.DirectoryServiceClient) *DirectoriesUploader {
+	return &DirectoriesUploader{
+		ctx:                       ctx,
+		directoryServiceClient:    directoryServiceClient,
+		directoryServicePutStream: nil,
+	}
+}
+
+func (du *DirectoriesUploader) Put(directory *storev1pb.Directory) error {
+	directoryDgst, err := directory.Digest()
+	if err != nil {
+		return fmt.Errorf("failed calculating directory digest: %w", err)
+	}
+
+	// Send the directory to the directory service
+	// If the stream hasn't been initialized yet, do it first
+	if du.directoryServicePutStream == nil {
+		directoryServicePutStream, err := du.directoryServiceClient.Put(du.ctx)
+		if err != nil {
+			return fmt.Errorf("unable to initialize directory service put stream: %v", err)
+		}
+		du.directoryServicePutStream = directoryServicePutStream
+	}
+
+	// send the directory out
+	err = du.directoryServicePutStream.Send(directory)
+	if err != nil {
+		return fmt.Errorf("error sending directory: %w", err)
+	}
+	log.WithField("digest", base64.StdEncoding.EncodeToString(directoryDgst)).Info("uploaded directory")
+
+	return nil
+}
+
+// Done is called whenever we're
+func (du *DirectoriesUploader) Done() (*storev1pb.PutDirectoryResponse, error) {
+	// only close once, and only if we opened.
+	if du.directoryServicePutStream == nil {
+		return nil, nil
+	}
+	putDirectoryResponse, err := du.directoryServicePutStream.CloseAndRecv()
+	if err != nil {
+		return nil, fmt.Errorf("unable to close directory service put stream: %v", err)
+	}
+
+	du.directoryServicePutStream = nil
+
+	return putDirectoryResponse, nil
+}
diff --git a/tvix/nar-bridge/pkg/server/nar_get.go b/tvix/nar-bridge/pkg/server/nar_get.go
new file mode 100644
index 000000000000..c8c35e69ff16
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/nar_get.go
@@ -0,0 +1,212 @@
+package server
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"net/http"
+	"sync"
+
+	"code.tvl.fyi/tvix/nar-bridge/pkg/writer"
+	storev1pb "code.tvl.fyi/tvix/store/protos"
+	"github.com/go-chi/chi/v5"
+	nixhash "github.com/nix-community/go-nix/pkg/hash"
+	"github.com/nix-community/go-nix/pkg/nixbase32"
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	narUrl = "/nar/{narhash:^([" + nixbase32.Alphabet + "]{52})$}.nar"
+)
+
+func renderNar(
+	ctx context.Context,
+	log *log.Entry,
+	directoryServiceClient storev1pb.DirectoryServiceClient,
+	blobServiceClient storev1pb.BlobServiceClient,
+	narHashToPathInfoMu *sync.Mutex,
+	narHashToPathInfo map[string]*storev1pb.PathInfo,
+	w io.Writer,
+	narHash *nixhash.Hash,
+	headOnly bool,
+) error {
+	// look in the lookup table
+	narHashToPathInfoMu.Lock()
+	pathInfo, found := narHashToPathInfo[narHash.SRIString()]
+	narHashToPathInfoMu.Unlock()
+
+	// if we didn't find anything, return 404.
+	if !found {
+		return fmt.Errorf("narHash not found: %w", fs.ErrNotExist)
+	}
+
+	// if this was only a head request, we're done.
+	if headOnly {
+		return nil
+	}
+
+	directories := make(map[string]*storev1pb.Directory)
+
+	// If the root node is a directory, ask the directory service for all directories
+	if pathInfoDirectory := pathInfo.GetNode().GetDirectory(); pathInfoDirectory != nil {
+		rootDirectoryDigest := pathInfoDirectory.GetDigest()
+		log = log.WithField("root_directory", base64.StdEncoding.EncodeToString(rootDirectoryDigest))
+
+		directoryStream, err := directoryServiceClient.Get(ctx, &storev1pb.GetDirectoryRequest{
+			ByWhat: &storev1pb.GetDirectoryRequest_Digest{
+				Digest: rootDirectoryDigest,
+			},
+			Recursive: true,
+		})
+		if err != nil {
+			return fmt.Errorf("unable to query directory stream: %w", err)
+		}
+
+		// For now, we just stream all of these locally and put them into a hashmap,
+		// which is used in the lookup function below.
+		for {
+			directory, err := directoryStream.Recv()
+			if err != nil {
+				if err == io.EOF {
+					break
+				}
+				return fmt.Errorf("unable to receive from directory stream: %w", err)
+			}
+
+			// calculate directory digest
+			// TODO: do we need to do any more validation?
+			directoryDgst, err := directory.Digest()
+			if err != nil {
+				return fmt.Errorf("unable to calculate directory digest: %w", err)
+			}
+
+			// TODO: debug level
+			log.WithField("directory", base64.StdEncoding.EncodeToString(directoryDgst)).Info("received directory node")
+
+			directories[hex.EncodeToString(directoryDgst)] = directory
+		}
+
+	}
+
+	// render the NAR file
+	err := writer.Export(
+		w,
+		pathInfo,
+		func(directoryDigest []byte) (*storev1pb.Directory, error) {
+			// TODO: debug level
+			log.WithField("directory", base64.StdEncoding.EncodeToString(directoryDigest)).Info("Get directory")
+			directoryRefStr := hex.EncodeToString(directoryDigest)
+			directory, found := directories[directoryRefStr]
+			if !found {
+				return nil, fmt.Errorf(
+					"directory with hash %v does not exist: %w",
+					directoryDigest,
+					fs.ErrNotExist,
+				)
+			}
+
+			return directory, nil
+		},
+		func(blobDigest []byte) (io.ReadCloser, error) {
+			// TODO: debug level
+			log.WithField("blob", base64.StdEncoding.EncodeToString(blobDigest)).Info("Get blob")
+			resp, err := blobServiceClient.Read(ctx, &storev1pb.ReadBlobRequest{
+				Digest: blobDigest,
+			})
+			if err != nil {
+				return nil, fmt.Errorf("unable to get blob: %w", err)
+
+			}
+
+			// TODO: spin up a goroutine producing this.
+			data := &bytes.Buffer{}
+			for {
+				chunk, err := resp.Recv()
+				if errors.Is(err, io.EOF) {
+					break
+				}
+				if err != nil {
+					return nil, fmt.Errorf("read chunk: %w", err)
+				}
+				_, err = data.Write(chunk.GetData())
+				if err != nil {
+					return nil, fmt.Errorf("buffer chunk: %w", err)
+				}
+			}
+			return io.NopCloser(data), nil
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("unable to export nar: %w", err)
+	}
+	return nil
+}
+
+func registerNarGet(s *Server) {
+	// TODO: properly compose this
+	s.handler.Head(narUrl, func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		ctx := r.Context()
+
+		// parse the narhash sent in the request URL
+		narHash, err := parseNarHashFromUrl(chi.URLParamFromCtx(ctx, "narhash"))
+		if err != nil {
+			log.WithError(err).WithField("url", r.URL).Error("unable to decode nar hash from url")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("unable to decode nar hash from url"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+
+		log := log.WithField("narhash_url", narHash.SRIString())
+
+		err = renderNar(ctx, log, s.directoryServiceClient, s.blobServiceClient, &s.narHashToPathInfoMu, s.narHashToPathInfo, w, narHash, true)
+		if err != nil {
+			log.WithError(err).Info("unable to render nar")
+			if errors.Is(err, fs.ErrNotExist) {
+				w.WriteHeader(http.StatusNotFound)
+			} else {
+				w.WriteHeader(http.StatusInternalServerError)
+			}
+		}
+
+	})
+	s.handler.Get(narUrl, func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		ctx := r.Context()
+
+		// parse the narhash sent in the request URL
+		narHash, err := parseNarHashFromUrl(chi.URLParamFromCtx(ctx, "narhash"))
+		if err != nil {
+			log.WithError(err).WithField("url", r.URL).Error("unable to decode nar hash from url")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("unable to decode nar hash from url"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+
+		log := log.WithField("narhash_url", narHash.SRIString())
+
+		err = renderNar(ctx, log, s.directoryServiceClient, s.blobServiceClient, &s.narHashToPathInfoMu, s.narHashToPathInfo, w, narHash, false)
+		if err != nil {
+			if errors.Is(err, fs.ErrNotExist) {
+				w.WriteHeader(http.StatusNotFound)
+			} else {
+				w.WriteHeader(http.StatusInternalServerError)
+			}
+		}
+	})
+}
diff --git a/tvix/nar-bridge/pkg/server/nar_put.go b/tvix/nar-bridge/pkg/server/nar_put.go
new file mode 100644
index 000000000000..9d6752e85bf1
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/nar_put.go
@@ -0,0 +1,140 @@
+package server
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	"net/http"
+
+	"code.tvl.fyi/tvix/nar-bridge/pkg/reader"
+	storev1pb "code.tvl.fyi/tvix/store/protos"
+	"github.com/go-chi/chi/v5"
+	nixhash "github.com/nix-community/go-nix/pkg/hash"
+	"github.com/nix-community/go-nix/pkg/nixbase32"
+	"github.com/sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
+)
+
+func registerNarPut(s *Server) {
+	s.handler.Put(narUrl, func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		ctx := r.Context()
+
+		// parse the narhash sent in the request URL
+		narHashFromUrl, err := parseNarHashFromUrl(chi.URLParamFromCtx(ctx, "narhash"))
+		if err != nil {
+			log.WithError(err).WithField("url", r.URL).Error("unable to decode nar hash from url")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("unable to decode nar hash from url"))
+			if err != nil {
+				log.WithError(err).Error("unable to write error message to client")
+			}
+
+			return
+		}
+
+		log := log.WithField("narhash_url", narHashFromUrl.SRIString())
+
+		directoriesUploader := NewDirectoriesUploader(ctx, s.directoryServiceClient)
+		defer directoriesUploader.Done() //nolint:errcheck
+
+		rd := reader.New(bufio.NewReader(r.Body))
+		pathInfo, err := rd.Import(
+			ctx,
+			genBlobServiceWriteCb(ctx, s.blobServiceClient),
+			func(directory *storev1pb.Directory) error {
+				return directoriesUploader.Put(directory)
+			},
+		)
+
+		if err != nil {
+			log.Errorf("error during NAR import: %v", err)
+			w.WriteHeader(http.StatusInternalServerError)
+			_, err := w.Write([]byte(fmt.Sprintf("error during NAR import: %v", err)))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+
+		log.Infof("closing the stream")
+
+		// Close the directories uploader
+		directoriesPutResponse, err := directoriesUploader.Done()
+		if err != nil {
+			log.WithError(err).Error("error during directory upload")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("error during directory upload"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+		// If we uploaded directories (so directoriesPutResponse doesn't return null),
+		// the RootDigest field in directoriesPutResponse should match the digest
+		// returned in the PathInfo struct returned by the `Import` call.
+		// This check ensures the server-side came up with the same root hash.
+
+		if directoriesPutResponse != nil {
+			rootDigestPathInfo := pathInfo.GetNode().GetDirectory().GetDigest()
+			rootDigestDirectoriesPutResponse := directoriesPutResponse.GetRootDigest()
+
+			log := log.WithFields(logrus.Fields{
+				"root_digest_pathinfo":             rootDigestPathInfo,
+				"root_digest_directories_put_resp": rootDigestDirectoriesPutResponse,
+			})
+			if !bytes.Equal(rootDigestPathInfo, rootDigestDirectoriesPutResponse) {
+				log.Errorf("returned root digest doesn't match what's calculated")
+
+				w.WriteHeader(http.StatusBadRequest)
+				_, err := w.Write([]byte("error in root digest calculation"))
+				if err != nil {
+					log.WithError(err).Error("unable to write error message to client")
+				}
+
+				return
+			}
+		}
+
+		// Compare the nar hash specified in the URL with the one that has been
+		// calculated while processing the NAR file
+		piNarHash, err := nixhash.ParseNixBase32(
+			"sha256:" + nixbase32.EncodeToString(pathInfo.GetNarinfo().NarSha256),
+		)
+		if err != nil {
+			panic("must parse nixbase32")
+		}
+
+		if !bytes.Equal(narHashFromUrl.Digest(), piNarHash.Digest()) {
+			log := log.WithFields(logrus.Fields{
+				"narhash_received_sha256": piNarHash.SRIString(),
+				"narsize":                 pathInfo.GetNarinfo().GetNarSize(),
+			})
+			log.Error("received bytes don't match narhash from URL")
+
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("received bytes don't match narHash specified in URL"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+
+		}
+
+		// Insert the partial pathinfo structs into our lookup map,
+		// so requesting the NAR file will be possible.
+		// The same  might exist already, but it'll have the same contents (so
+		// replacing will be a no-op), except maybe the root node Name field value, which
+		// is safe to ignore (as not part of the NAR).
+		s.narHashToPathInfoMu.Lock()
+		s.narHashToPathInfo[piNarHash.SRIString()] = pathInfo
+		s.narHashToPathInfoMu.Unlock()
+
+		// Done!
+	})
+
+}
diff --git a/tvix/nar-bridge/pkg/server/narinfo_get.go b/tvix/nar-bridge/pkg/server/narinfo_get.go
new file mode 100644
index 000000000000..977e1136130f
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/narinfo_get.go
@@ -0,0 +1,146 @@
+package server
+
+import (
+	"context"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"io"
+	"io/fs"
+	"net/http"
+	"path"
+	"strings"
+	"sync"
+
+	storev1pb "code.tvl.fyi/tvix/store/protos"
+	"github.com/go-chi/chi/v5"
+	nixhash "github.com/nix-community/go-nix/pkg/hash"
+	"github.com/nix-community/go-nix/pkg/narinfo"
+	"github.com/nix-community/go-nix/pkg/narinfo/signature"
+	"github.com/nix-community/go-nix/pkg/nixbase32"
+	"github.com/nix-community/go-nix/pkg/nixpath"
+	log "github.com/sirupsen/logrus"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// renderNarinfo writes narinfo contents to a passes io.Writer, or a returns a
+// (wrapped) io.ErrNoExist error if something doesn't exist.
+// if headOnly is set to true, only the existence is checked, but no content is
+// actually written.
+func renderNarinfo(
+	ctx context.Context,
+	log *log.Entry,
+	pathInfoServiceClient storev1pb.PathInfoServiceClient,
+	narHashToPathInfoMu *sync.Mutex,
+	narHashToPathInfo map[string]*storev1pb.PathInfo,
+	outputHash []byte,
+	w io.Writer,
+	headOnly bool,
+) error {
+	pathInfo, err := pathInfoServiceClient.Get(ctx, &storev1pb.GetPathInfoRequest{
+		ByWhat: &storev1pb.GetPathInfoRequest_ByOutputHash{
+			ByOutputHash: outputHash,
+		},
+	})
+	if err != nil {
+		st, ok := status.FromError(err)
+		if ok {
+			if st.Code() == codes.NotFound {
+				return fmt.Errorf("output hash %v not found: %w", base64.StdEncoding.EncodeToString(outputHash), fs.ErrNotExist)
+			}
+			return fmt.Errorf("unable to get pathinfo, code %v: %w", st.Code(), err)
+		}
+
+		return fmt.Errorf("unable to get pathinfo: %w", err)
+	}
+
+	narHash, err := nixhash.ParseNixBase32("sha256:" + nixbase32.EncodeToString(pathInfo.GetNarinfo().GetNarSha256()))
+	if err != nil {
+		// TODO: return proper error
+		return fmt.Errorf("No usable NarHash found in PathInfo")
+	}
+
+	// add things to the lookup table, in case the same process didn't handle the NAR hash yet.
+	narHashToPathInfoMu.Lock()
+	narHashToPathInfo[narHash.SRIString()] = pathInfo
+	narHashToPathInfoMu.Unlock()
+
+	if headOnly {
+		return nil
+	}
+
+	// convert the signatures from storev1pb signatures to narinfo signatures
+	narinfoSignatures := make([]signature.Signature, 0)
+	for _, pathInfoSignature := range pathInfo.Narinfo.Signatures {
+		narinfoSignatures = append(narinfoSignatures, signature.Signature{
+			Name: pathInfoSignature.GetName(),
+			Data: pathInfoSignature.GetData(),
+		})
+	}
+
+	// extract the name of the node in the pathInfo structure, which will become the output path
+	var nodeName []byte
+	switch v := (pathInfo.GetNode().GetNode()).(type) {
+	case *storev1pb.Node_File:
+		nodeName = v.File.GetName()
+	case *storev1pb.Node_Symlink:
+		nodeName = v.Symlink.GetName()
+	case *storev1pb.Node_Directory:
+		nodeName = v.Directory.GetName()
+	}
+
+	narInfo := narinfo.NarInfo{
+		StorePath:   path.Join(nixpath.StoreDir, string(nodeName)),
+		URL:         "nar/" + nixbase32.EncodeToString(narHash.Digest()) + ".nar",
+		Compression: "none", // TODO: implement zstd compression
+		NarHash:     narHash,
+		NarSize:     uint64(pathInfo.Narinfo.NarSize),
+		References:  pathInfo.Narinfo.GetReferenceNames(),
+		Signatures:  narinfoSignatures,
+	}
+
+	// render .narinfo from pathInfo
+	_, err = io.Copy(w, strings.NewReader(narInfo.String()))
+	if err != nil {
+		return fmt.Errorf("unable to write narinfo to client: %w", err)
+	}
+
+	return nil
+}
+
+func registerNarinfoGet(s *Server) {
+	// GET $outHash.narinfo looks up the PathInfo from the tvix-store,
+	// and then render a .narinfo file to the client.
+	// It will keep the PathInfo in the lookup map,
+	// so a subsequent GET /nar/ $narhash.nar request can find it.
+	s.handler.Get("/{outputhash:^["+nixbase32.Alphabet+"]{32}}.narinfo", func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		ctx := r.Context()
+		log := log.WithField("outputhash", chi.URLParamFromCtx(ctx, "outputhash"))
+
+		// parse the output hash sent in the request URL
+		outputHash, err := nixbase32.DecodeString(chi.URLParamFromCtx(ctx, "outputhash"))
+		if err != nil {
+			log.WithError(err).Error("unable to decode output hash from url")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("unable to decode output hash from url"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+
+		err = renderNarinfo(ctx, log, s.pathInfoServiceClient, &s.narHashToPathInfoMu, s.narHashToPathInfo, outputHash, w, false)
+		if err != nil {
+			log.WithError(err).Info("unable to render narinfo")
+			if errors.Is(err, fs.ErrNotExist) {
+				w.WriteHeader(http.StatusNotFound)
+			} else {
+				w.WriteHeader(http.StatusInternalServerError)
+			}
+		}
+	})
+}
diff --git a/tvix/nar-bridge/pkg/server/narinfo_put.go b/tvix/nar-bridge/pkg/server/narinfo_put.go
new file mode 100644
index 000000000000..c5b4094f8582
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/narinfo_put.go
@@ -0,0 +1,174 @@
+package server
+
+import (
+	"net/http"
+	"path"
+
+	storev1pb "code.tvl.fyi/tvix/store/protos"
+	"github.com/go-chi/chi/v5"
+	"github.com/nix-community/go-nix/pkg/narinfo"
+	"github.com/nix-community/go-nix/pkg/nixbase32"
+	"github.com/nix-community/go-nix/pkg/nixpath"
+	"github.com/sirupsen/logrus"
+	log "github.com/sirupsen/logrus"
+)
+
+func registerNarinfoPut(s *Server) {
+	s.handler.Put("/{outputhash:^["+nixbase32.Alphabet+"]{32}}.narinfo", func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		ctx := r.Context()
+		log := log.WithField("outputhash", chi.URLParamFromCtx(ctx, "outputhash"))
+
+		// TODO: decide on merging behaviour.
+		// Maybe it's fine to add if contents are the same, but more sigs can be added?
+		// Right now, just replace a .narinfo for a path that already exists.
+
+		// read and parse the .narinfo file
+		narInfo, err := narinfo.Parse(r.Body)
+		if err != nil {
+			log.WithError(err).Error("unable to parse narinfo")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("unable to parse narinfo"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+
+		log = log.WithFields(logrus.Fields{
+			"narhash":     narInfo.NarHash.SRIString(),
+			"output_path": narInfo.StorePath,
+		})
+
+		var pathInfo *storev1pb.PathInfo
+
+		// look up the narHash in our temporary map
+		s.narHashToPathInfoMu.Lock()
+		pathInfo, found := s.narHashToPathInfo[narInfo.NarHash.SRIString()]
+		s.narHashToPathInfoMu.Unlock()
+		if !found {
+			log.Error("unable to find referred NAR")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("unable to find referred NAR"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+
+		// compare fields with what we computed while receiving the NAR file
+
+		// NarSize needs to match
+		if pathInfo.Narinfo.NarSize != narInfo.NarSize {
+			log.Error("narsize mismatch")
+			w.WriteHeader(http.StatusBadRequest)
+			_, err := w.Write([]byte("unable to parse narinfo"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+		// We know the narhash in the .narinfo matches one of the two narhashes in the partial pathInfo,
+		// because that's how we found it.
+
+		// FUTUREWORK: We can't compare References yet, but it'd be a good idea to
+		// do reference checking on .nar files server-side during upload.
+		// We however still need to be parse them, because we store
+		// the bytes in pathInfo.References, and the full strings in pathInfo.Narinfo.ReferenceNames.
+		referencesBytes := make([][]byte, 0)
+		for _, reference := range narInfo.References {
+			np, err := nixpath.FromString(path.Join(nixpath.StoreDir, reference))
+			if err != nil {
+				log.WithField("reference", reference).WithError(err).Error("unable to parse reference")
+				w.WriteHeader(http.StatusBadRequest)
+				_, err := w.Write([]byte("unable to parse reference"))
+				if err != nil {
+					log.WithError(err).Errorf("unable to write error message to client")
+				}
+
+				return
+			}
+			referencesBytes = append(referencesBytes, np.Digest)
+		}
+
+		// assemble the []*storev1pb.NARInfo_Signature{} from narinfo.Signatures.
+		pbNarinfoSignatures := make([]*storev1pb.NARInfo_Signature, 0)
+		for _, narinfoSig := range narInfo.Signatures {
+
+			pbNarinfoSignatures = append(pbNarinfoSignatures, &storev1pb.NARInfo_Signature{
+				Name: narinfoSig.Name,
+				Data: narinfoSig.Data,
+			})
+		}
+
+		// If everything matches, We will add References, NAR signatures and the
+		// output path name, and then upload to the pathinfo service.
+		// We want a copy here, because we don't want to mutate the contents in the lookup table
+		// until we get things back from the remote store.
+		pathInfoToUpload := &storev1pb.PathInfo{
+			Node:       nil, // set below
+			References: referencesBytes,
+			Narinfo: &storev1pb.NARInfo{
+				NarSize:        pathInfo.Narinfo.NarSize,
+				NarSha256:      pathInfo.Narinfo.NarSha256,
+				Signatures:     pbNarinfoSignatures,
+				ReferenceNames: narInfo.References,
+			},
+		}
+
+		// We need to add the basename of the storepath from the .narinfo
+		// to the pathInfo to be sent.
+		switch v := (pathInfo.GetNode().GetNode()).(type) {
+		case *storev1pb.Node_File:
+			pathInfoToUpload.Node = &storev1pb.Node{
+				Node: &storev1pb.Node_File{
+					File: &storev1pb.FileNode{
+						Name:       []byte(path.Base(narInfo.StorePath)),
+						Digest:     v.File.Digest,
+						Size:       v.File.Size,
+						Executable: v.File.Executable,
+					},
+				},
+			}
+		case *storev1pb.Node_Symlink:
+			pathInfoToUpload.Node = &storev1pb.Node{
+				Node: &storev1pb.Node_Symlink{
+					Symlink: &storev1pb.SymlinkNode{
+						Name:   []byte(path.Base(narInfo.StorePath)),
+						Target: v.Symlink.Target,
+					},
+				},
+			}
+		case *storev1pb.Node_Directory:
+			pathInfoToUpload.Node = &storev1pb.Node{
+				Node: &storev1pb.Node_Directory{
+					Directory: &storev1pb.DirectoryNode{
+						Name:   []byte(path.Base(narInfo.StorePath)),
+						Digest: v.Directory.Digest,
+						Size:   v.Directory.Size,
+					},
+				},
+			}
+		}
+
+		receivedPathInfo, err := s.pathInfoServiceClient.Put(ctx, pathInfoToUpload)
+		if err != nil {
+			log.WithError(err).Error("unable to upload pathinfo to service")
+			w.WriteHeader(http.StatusInternalServerError)
+			_, err := w.Write([]byte("unable to upload pathinfo to server"))
+			if err != nil {
+				log.WithError(err).Errorf("unable to write error message to client")
+			}
+
+			return
+		}
+
+		log.Infof("received new pathInfo: %v+", receivedPathInfo)
+
+		// TODO: update the local temporary pathinfo with this?
+	})
+}
diff --git a/tvix/nar-bridge/pkg/server/server.go b/tvix/nar-bridge/pkg/server/server.go
new file mode 100644
index 000000000000..083b7f295a12
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/server.go
@@ -0,0 +1,86 @@
+package server
+
+import (
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	storev1pb "code.tvl.fyi/tvix/store/protos"
+	"github.com/go-chi/chi/middleware"
+	"github.com/go-chi/chi/v5"
+	log "github.com/sirupsen/logrus"
+)
+
+type Server struct {
+	handler chi.Router
+
+	directoryServiceClient storev1pb.DirectoryServiceClient
+	blobServiceClient      storev1pb.BlobServiceClient
+	pathInfoServiceClient  storev1pb.PathInfoServiceClient
+
+	// When uploading NAR files to a HTTP binary cache, the .nar
+	// files are uploaded before the .narinfo files.
+	// We need *both* to be able to fully construct a PathInfo object.
+	// Keep a in-memory map of narhash(es) (in SRI) to sparse PathInfo.
+	// This is necessary until we can ask a PathInfoService for a node with a given
+	// narSha256.
+	narHashToPathInfoMu sync.Mutex
+	narHashToPathInfo   map[string]*storev1pb.PathInfo
+}
+
+func New(
+	directoryServiceClient storev1pb.DirectoryServiceClient,
+	blobServiceClient storev1pb.BlobServiceClient,
+	pathInfoServiceClient storev1pb.PathInfoServiceClient,
+	enableAccessLog bool,
+	priority int,
+) *Server {
+	r := chi.NewRouter()
+
+	if enableAccessLog {
+		r.Use(middleware.Logger)
+	}
+
+	r.Get("/", func(w http.ResponseWriter, r *http.Request) {
+		_, err := w.Write([]byte("nar-bridge"))
+		if err != nil {
+			log.Errorf("Unable to write response: %v", err)
+		}
+	})
+
+	r.Get("/nix-cache-info", func(w http.ResponseWriter, r *http.Request) {
+		_, err := w.Write([]byte(fmt.Sprintf("StoreDir: /nix/store\nWantMassQuery: 1\nPriority: %d\n", priority)))
+		if err != nil {
+			log.Errorf("Unable to write response: %v", err)
+		}
+	})
+
+	s := &Server{
+		handler:                r,
+		directoryServiceClient: directoryServiceClient,
+		blobServiceClient:      blobServiceClient,
+		pathInfoServiceClient:  pathInfoServiceClient,
+		narHashToPathInfo:      make(map[string]*storev1pb.PathInfo),
+	}
+
+	registerNarPut(s)
+	registerNarinfoPut(s)
+
+	registerNarinfoGet(s)
+	registerNarGet(s)
+
+	return s
+}
+
+func (s *Server) ListenAndServe(addr string) error {
+	srv := &http.Server{
+		Addr:         addr,
+		Handler:      s.handler,
+		ReadTimeout:  50 * time.Second,
+		WriteTimeout: 100 * time.Second,
+		IdleTimeout:  150 * time.Second,
+	}
+
+	return srv.ListenAndServe()
+}
diff --git a/tvix/nar-bridge/pkg/server/util.go b/tvix/nar-bridge/pkg/server/util.go
new file mode 100644
index 000000000000..47e368adde80
--- /dev/null
+++ b/tvix/nar-bridge/pkg/server/util.go
@@ -0,0 +1,24 @@
+package server
+
+import (
+	"fmt"
+	nixhash "github.com/nix-community/go-nix/pkg/hash"
+)
+
+// parseNarHashFromUrl parses a nixbase32 string representing a sha256 NarHash
+// and returns a nixhash.Hash when it was able to parse, or an error.
+func parseNarHashFromUrl(narHashFromUrl string) (*nixhash.Hash, error) {
+	// peek at the length. If it's 52 characters, assume sha256,
+	// if it's something else, this is an error.
+	l := len(narHashFromUrl)
+	if l != 52 {
+		return nil, fmt.Errorf("invalid length of narHash: %v", l)
+	}
+
+	nixHash, err := nixhash.ParseNixBase32("sha256:" + narHashFromUrl)
+	if err != nil {
+		return nil, fmt.Errorf("unable to parse nixbase32 hash: %w", err)
+	}
+
+	return nixHash, nil
+}
generated by cgit-pink 1.4.1 (git 2.44.2) at 2024-11-30T16ยท38+0000