From 28d1b9c01d009424eed276f5689430897afd97ec Mon Sep 17 00:00:00 2001 From: Florian Klink Date: Tue, 3 Oct 2023 14:49:18 +0300 Subject: refactor(tvix/nar-bridge): move pkg/server to pkg/http This is only dealing with the HTTP interface. Change-Id: I011b624fd9f11ea96231b92fea1166c118a219f2 Reviewed-on: https://cl.tvl.fyi/c/depot/+/9535 Tested-by: BuildkiteCI Autosubmit: flokli Reviewed-by: Connor Brewster --- tvix/nar-bridge/cmd/nar-bridge/main.go | 4 +- tvix/nar-bridge/pkg/http/nar_get.go | 219 ++++++++++++++++++++++++++++++ tvix/nar-bridge/pkg/http/nar_put.go | 141 +++++++++++++++++++ tvix/nar-bridge/pkg/http/narinfo_get.go | 147 ++++++++++++++++++++ tvix/nar-bridge/pkg/http/narinfo_put.go | 175 ++++++++++++++++++++++++ tvix/nar-bridge/pkg/http/server.go | 95 +++++++++++++ tvix/nar-bridge/pkg/http/util.go | 24 ++++ tvix/nar-bridge/pkg/server/nar_get.go | 219 ------------------------------ tvix/nar-bridge/pkg/server/nar_put.go | 141 ------------------- tvix/nar-bridge/pkg/server/narinfo_get.go | 147 -------------------- tvix/nar-bridge/pkg/server/narinfo_put.go | 175 ------------------------ tvix/nar-bridge/pkg/server/server.go | 95 ------------- tvix/nar-bridge/pkg/server/util.go | 24 ---- 13 files changed, 803 insertions(+), 803 deletions(-) create mode 100644 tvix/nar-bridge/pkg/http/nar_get.go create mode 100644 tvix/nar-bridge/pkg/http/nar_put.go create mode 100644 tvix/nar-bridge/pkg/http/narinfo_get.go create mode 100644 tvix/nar-bridge/pkg/http/narinfo_put.go create mode 100644 tvix/nar-bridge/pkg/http/server.go create mode 100644 tvix/nar-bridge/pkg/http/util.go delete mode 100644 tvix/nar-bridge/pkg/server/nar_get.go delete mode 100644 tvix/nar-bridge/pkg/server/nar_put.go delete mode 100644 tvix/nar-bridge/pkg/server/narinfo_get.go delete mode 100644 tvix/nar-bridge/pkg/server/narinfo_put.go delete mode 100644 tvix/nar-bridge/pkg/server/server.go delete mode 100644 tvix/nar-bridge/pkg/server/util.go (limited to 'tvix') diff --git a/tvix/nar-bridge/cmd/nar-bridge/main.go b/tvix/nar-bridge/cmd/nar-bridge/main.go index 0f7445ced5..d86e84b870 100644 --- a/tvix/nar-bridge/cmd/nar-bridge/main.go +++ b/tvix/nar-bridge/cmd/nar-bridge/main.go @@ -12,7 +12,7 @@ import ( "google.golang.org/grpc/credentials/insecure" castorev1pb "code.tvl.fyi/tvix/castore/protos" - "code.tvl.fyi/tvix/nar-bridge/pkg/server" + narBridgeHttp "code.tvl.fyi/tvix/nar-bridge/pkg/http" storev1pb "code.tvl.fyi/tvix/store/protos" "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus" @@ -47,7 +47,7 @@ func main() { } defer conn.Close() - s := server.New( + s := narBridgeHttp.New( castorev1pb.NewDirectoryServiceClient(conn), castorev1pb.NewBlobServiceClient(conn), storev1pb.NewPathInfoServiceClient(conn), diff --git a/tvix/nar-bridge/pkg/http/nar_get.go b/tvix/nar-bridge/pkg/http/nar_get.go new file mode 100644 index 0000000000..be7eb82606 --- /dev/null +++ b/tvix/nar-bridge/pkg/http/nar_get.go @@ -0,0 +1,219 @@ +package http + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/hex" + "errors" + "fmt" + "io" + "io/fs" + "net/http" + "sync" + + castorev1pb "code.tvl.fyi/tvix/castore/protos" + "code.tvl.fyi/tvix/nar-bridge/pkg/exporter" + 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 castorev1pb.DirectoryServiceClient, + blobServiceClient castorev1pb.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]*castorev1pb.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, &castorev1pb.GetDirectoryRequest{ + ByWhat: &castorev1pb.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) + } + + log.WithField("directory", base64.StdEncoding.EncodeToString(directoryDgst)).Debug("received directory node") + + directories[hex.EncodeToString(directoryDgst)] = directory + } + + } + + // render the NAR file + err := exporter.Export( + w, + pathInfo, + func(directoryDigest []byte) (*castorev1pb.Directory, error) { + log.WithField("directory", base64.StdEncoding.EncodeToString(directoryDigest)).Debug("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) { + log.WithField("blob", base64.StdEncoding.EncodeToString(blobDigest)).Debug("Get blob") + resp, err := blobServiceClient.Read(ctx, &castorev1pb.ReadBlobRequest{ + Digest: blobDigest, + }) + if err != nil { + return nil, fmt.Errorf("unable to get blob: %w", err) + } + + // set up a pipe, let a goroutine write, return the reader. + pR, pW := io.Pipe() + + go func() { + for { + chunk, err := resp.Recv() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + pW.CloseWithError(fmt.Errorf("receiving chunk: %w", err)) + return + } + + // write the received chunk to the writer part of the pipe + if _, err := io.Copy(pW, bytes.NewReader(chunk.GetData())); err != nil { + log.WithError(err).Error("writing chunk to pipe") + pW.CloseWithError(fmt.Errorf("writing chunk to pipe: %w", err)) + return + } + } + pW.Close() + + }() + + return io.NopCloser(pR), 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 { + if errors.Is(err, fs.ErrNotExist) { + w.WriteHeader(http.StatusNotFound) + } else { + log.WithError(err).Warn("unable to render nar") + 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/http/nar_put.go b/tvix/nar-bridge/pkg/http/nar_put.go new file mode 100644 index 0000000000..52ab425dfc --- /dev/null +++ b/tvix/nar-bridge/pkg/http/nar_put.go @@ -0,0 +1,141 @@ +package http + +import ( + "bufio" + "bytes" + "fmt" + "net/http" + + castorev1pb "code.tvl.fyi/tvix/castore/protos" + "code.tvl.fyi/tvix/nar-bridge/pkg/importer" + "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 := importer.NewDirectoriesUploader(ctx, s.directoryServiceClient) + defer directoriesUploader.Done() //nolint:errcheck + + pathInfo, err := importer.Import( + ctx, + // buffer the body by 10MiB + bufio.NewReaderSize(r.Body, 10*1024*1024), + importer.GenBlobUploaderCb(ctx, s.blobServiceClient), + func(directory *castorev1pb.Directory) ([]byte, 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.Debug("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/http/narinfo_get.go b/tvix/nar-bridge/pkg/http/narinfo_get.go new file mode 100644 index 0000000000..91fe6c5603 --- /dev/null +++ b/tvix/nar-bridge/pkg/http/narinfo_get.go @@ -0,0 +1,147 @@ +package http + +import ( + "context" + "encoding/base64" + "errors" + "fmt" + "io" + "io/fs" + "net/http" + "path" + "strings" + "sync" + + castorev1pb "code.tvl.fyi/tvix/castore/protos" + 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 *castorev1pb.Node_File: + nodeName = v.File.GetName() + case *castorev1pb.Node_Symlink: + nodeName = v.Symlink.GetName() + case *castorev1pb.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 { + if errors.Is(err, fs.ErrNotExist) { + w.WriteHeader(http.StatusNotFound) + } else { + log.WithError(err).Warn("unable to render narinfo") + w.WriteHeader(http.StatusInternalServerError) + } + } + }) +} diff --git a/tvix/nar-bridge/pkg/http/narinfo_put.go b/tvix/nar-bridge/pkg/http/narinfo_put.go new file mode 100644 index 0000000000..831f93fbdf --- /dev/null +++ b/tvix/nar-bridge/pkg/http/narinfo_put.go @@ -0,0 +1,175 @@ +package http + +import ( + "net/http" + "path" + + castorev1pb "code.tvl.fyi/tvix/castore/protos" + 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 *castorev1pb.Node_File: + pathInfoToUpload.Node = &castorev1pb.Node{ + Node: &castorev1pb.Node_File{ + File: &castorev1pb.FileNode{ + Name: []byte(path.Base(narInfo.StorePath)), + Digest: v.File.Digest, + Size: v.File.Size, + Executable: v.File.Executable, + }, + }, + } + case *castorev1pb.Node_Symlink: + pathInfoToUpload.Node = &castorev1pb.Node{ + Node: &castorev1pb.Node_Symlink{ + Symlink: &castorev1pb.SymlinkNode{ + Name: []byte(path.Base(narInfo.StorePath)), + Target: v.Symlink.Target, + }, + }, + } + case *castorev1pb.Node_Directory: + pathInfoToUpload.Node = &castorev1pb.Node{ + Node: &castorev1pb.Node_Directory{ + Directory: &castorev1pb.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.Debugf("received new pathInfo: %v+", receivedPathInfo) + + // TODO: update the local temporary pathinfo with this? + }) +} diff --git a/tvix/nar-bridge/pkg/http/server.go b/tvix/nar-bridge/pkg/http/server.go new file mode 100644 index 0000000000..e8eb229ab4 --- /dev/null +++ b/tvix/nar-bridge/pkg/http/server.go @@ -0,0 +1,95 @@ +package http + +import ( + "context" + "fmt" + "net/http" + "sync" + "time" + + castorev1pb "code.tvl.fyi/tvix/castore/protos" + 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 { + srv *http.Server + handler chi.Router + + directoryServiceClient castorev1pb.DirectoryServiceClient + blobServiceClient castorev1pb.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 castorev1pb.DirectoryServiceClient, + blobServiceClient castorev1pb.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) Shutdown(ctx context.Context) error { + return s.srv.Shutdown(ctx) +} + +// ListenAndServer starts the webserver, and waits for it being closed or +// shutdown, after which it'll return ErrServerClosed. +func (s *Server) ListenAndServe(addr string) error { + s.srv = &http.Server{ + Addr: addr, + Handler: s.handler, + ReadTimeout: 500 * time.Second, + WriteTimeout: 500 * time.Second, + IdleTimeout: 500 * time.Second, + } + + return s.srv.ListenAndServe() +} diff --git a/tvix/nar-bridge/pkg/http/util.go b/tvix/nar-bridge/pkg/http/util.go new file mode 100644 index 0000000000..60febea1f4 --- /dev/null +++ b/tvix/nar-bridge/pkg/http/util.go @@ -0,0 +1,24 @@ +package http + +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 +} diff --git a/tvix/nar-bridge/pkg/server/nar_get.go b/tvix/nar-bridge/pkg/server/nar_get.go deleted file mode 100644 index 3ccb8d658d..0000000000 --- a/tvix/nar-bridge/pkg/server/nar_get.go +++ /dev/null @@ -1,219 +0,0 @@ -package server - -import ( - "bytes" - "context" - "encoding/base64" - "encoding/hex" - "errors" - "fmt" - "io" - "io/fs" - "net/http" - "sync" - - castorev1pb "code.tvl.fyi/tvix/castore/protos" - "code.tvl.fyi/tvix/nar-bridge/pkg/exporter" - 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 castorev1pb.DirectoryServiceClient, - blobServiceClient castorev1pb.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]*castorev1pb.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, &castorev1pb.GetDirectoryRequest{ - ByWhat: &castorev1pb.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) - } - - log.WithField("directory", base64.StdEncoding.EncodeToString(directoryDgst)).Debug("received directory node") - - directories[hex.EncodeToString(directoryDgst)] = directory - } - - } - - // render the NAR file - err := exporter.Export( - w, - pathInfo, - func(directoryDigest []byte) (*castorev1pb.Directory, error) { - log.WithField("directory", base64.StdEncoding.EncodeToString(directoryDigest)).Debug("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) { - log.WithField("blob", base64.StdEncoding.EncodeToString(blobDigest)).Debug("Get blob") - resp, err := blobServiceClient.Read(ctx, &castorev1pb.ReadBlobRequest{ - Digest: blobDigest, - }) - if err != nil { - return nil, fmt.Errorf("unable to get blob: %w", err) - } - - // set up a pipe, let a goroutine write, return the reader. - pR, pW := io.Pipe() - - go func() { - for { - chunk, err := resp.Recv() - if errors.Is(err, io.EOF) { - break - } - if err != nil { - pW.CloseWithError(fmt.Errorf("receiving chunk: %w", err)) - return - } - - // write the received chunk to the writer part of the pipe - if _, err := io.Copy(pW, bytes.NewReader(chunk.GetData())); err != nil { - log.WithError(err).Error("writing chunk to pipe") - pW.CloseWithError(fmt.Errorf("writing chunk to pipe: %w", err)) - return - } - } - pW.Close() - - }() - - return io.NopCloser(pR), 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 { - if errors.Is(err, fs.ErrNotExist) { - w.WriteHeader(http.StatusNotFound) - } else { - log.WithError(err).Warn("unable to render nar") - 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 deleted file mode 100644 index 41cd257f72..0000000000 --- a/tvix/nar-bridge/pkg/server/nar_put.go +++ /dev/null @@ -1,141 +0,0 @@ -package server - -import ( - "bufio" - "bytes" - "fmt" - "net/http" - - castorev1pb "code.tvl.fyi/tvix/castore/protos" - "code.tvl.fyi/tvix/nar-bridge/pkg/importer" - "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 := importer.NewDirectoriesUploader(ctx, s.directoryServiceClient) - defer directoriesUploader.Done() //nolint:errcheck - - pathInfo, err := importer.Import( - ctx, - // buffer the body by 10MiB - bufio.NewReaderSize(r.Body, 10*1024*1024), - importer.GenBlobUploaderCb(ctx, s.blobServiceClient), - func(directory *castorev1pb.Directory) ([]byte, 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.Debug("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 deleted file mode 100644 index f2276400c6..0000000000 --- a/tvix/nar-bridge/pkg/server/narinfo_get.go +++ /dev/null @@ -1,147 +0,0 @@ -package server - -import ( - "context" - "encoding/base64" - "errors" - "fmt" - "io" - "io/fs" - "net/http" - "path" - "strings" - "sync" - - castorev1pb "code.tvl.fyi/tvix/castore/protos" - 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 *castorev1pb.Node_File: - nodeName = v.File.GetName() - case *castorev1pb.Node_Symlink: - nodeName = v.Symlink.GetName() - case *castorev1pb.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 { - if errors.Is(err, fs.ErrNotExist) { - w.WriteHeader(http.StatusNotFound) - } else { - log.WithError(err).Warn("unable to render narinfo") - w.WriteHeader(http.StatusInternalServerError) - } - } - }) -} diff --git a/tvix/nar-bridge/pkg/server/narinfo_put.go b/tvix/nar-bridge/pkg/server/narinfo_put.go deleted file mode 100644 index 55543c9bc2..0000000000 --- a/tvix/nar-bridge/pkg/server/narinfo_put.go +++ /dev/null @@ -1,175 +0,0 @@ -package server - -import ( - "net/http" - "path" - - castorev1pb "code.tvl.fyi/tvix/castore/protos" - 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 *castorev1pb.Node_File: - pathInfoToUpload.Node = &castorev1pb.Node{ - Node: &castorev1pb.Node_File{ - File: &castorev1pb.FileNode{ - Name: []byte(path.Base(narInfo.StorePath)), - Digest: v.File.Digest, - Size: v.File.Size, - Executable: v.File.Executable, - }, - }, - } - case *castorev1pb.Node_Symlink: - pathInfoToUpload.Node = &castorev1pb.Node{ - Node: &castorev1pb.Node_Symlink{ - Symlink: &castorev1pb.SymlinkNode{ - Name: []byte(path.Base(narInfo.StorePath)), - Target: v.Symlink.Target, - }, - }, - } - case *castorev1pb.Node_Directory: - pathInfoToUpload.Node = &castorev1pb.Node{ - Node: &castorev1pb.Node_Directory{ - Directory: &castorev1pb.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.Debugf("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 deleted file mode 100644 index e50c637541..0000000000 --- a/tvix/nar-bridge/pkg/server/server.go +++ /dev/null @@ -1,95 +0,0 @@ -package server - -import ( - "context" - "fmt" - "net/http" - "sync" - "time" - - castorev1pb "code.tvl.fyi/tvix/castore/protos" - 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 { - srv *http.Server - handler chi.Router - - directoryServiceClient castorev1pb.DirectoryServiceClient - blobServiceClient castorev1pb.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 castorev1pb.DirectoryServiceClient, - blobServiceClient castorev1pb.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) Shutdown(ctx context.Context) error { - return s.srv.Shutdown(ctx) -} - -// ListenAndServer starts the webserver, and waits for it being closed or -// shutdown, after which it'll return ErrServerClosed. -func (s *Server) ListenAndServe(addr string) error { - s.srv = &http.Server{ - Addr: addr, - Handler: s.handler, - ReadTimeout: 500 * time.Second, - WriteTimeout: 500 * time.Second, - IdleTimeout: 500 * time.Second, - } - - return s.srv.ListenAndServe() -} diff --git a/tvix/nar-bridge/pkg/server/util.go b/tvix/nar-bridge/pkg/server/util.go deleted file mode 100644 index 47e368adde..0000000000 --- a/tvix/nar-bridge/pkg/server/util.go +++ /dev/null @@ -1,24 +0,0 @@ -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 -} -- cgit 1.4.1