diff --git a/knotmirror/resyncer.go b/knotmirror/resyncer.go --- a/knotmirror/resyncer.go +++ b/knotmirror/resyncer.go @@ -1,10 +1,12 @@ package knotmirror import ( + "bytes" "context" "database/sql" "errors" "fmt" + "io" "log/slog" "math/rand" "net/http" @@ -229,7 +231,8 @@ // HACK: check knot reachability with short timeout before running actual fetch. // This is crucial as git-cli doesn't support http connection timeout. // `http.lowSpeedTime` is only applied _after_ the connection. - if err := r.checkKnotReachability(ctx, repo); err != nil { + format, err := r.checkKnot(ctx, repo) + if err != nil { if isRateLimitError(err) { r.knotBackoffMu.Lock() r.knotBackoff[repo.KnotDomain] = time.Now().Add(10 * time.Second) @@ -238,6 +241,10 @@ } // TODO: suspend repo on 404. KnotStream updates will change the repo state back online return false, fmt.Errorf("knot unreachable: %w", err) + } + + if format == models.ObjectFormatSHA256 { + return r.suspendUnsupported(ctx, repo) } timeout := r.repoFetchTimeout @@ -282,11 +289,10 @@ return false } -// checkKnotReachability checks if Knot is reachable and is valid git remote server -func (r *Resyncer) checkKnotReachability(ctx context.Context, repo *models.Repo) error { +func (r *Resyncer) checkKnot(ctx context.Context, repo *models.Repo) (models.ObjectFormat, error) { repoUrl, err := makeRepoRemoteUrl(repo.KnotDomain, repo.RepoIdentifier(), r.cfg.KnotUseSSL) if err != nil { - return err + return "", err } repoUrl += "/info/refs?service=git-upload-pack" @@ -295,7 +301,7 @@ req, err := http.NewRequestWithContext(ctx, "GET", repoUrl, nil) if err != nil { - return err + return "", err } req.Header.Set("User-Agent", "git/2.x") req.Header.Set("Accept", "*/*") @@ -304,23 +310,48 @@ if err != nil { var uerr *url.Error if errors.As(err, &uerr) { - return fmt.Errorf("request failed: %w", uerr.Unwrap()) + return "", fmt.Errorf("request failed: %w", uerr.Unwrap()) } - return fmt.Errorf("request failed: %w", err) + return "", fmt.Errorf("request failed: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return &knotStatusError{resp.StatusCode} + return "", &knotStatusError{resp.StatusCode} } // check if target is git server ct := resp.Header.Get("Content-Type") if !strings.Contains(ct, "application/x-git-upload-pack-advertisement") { - return fmt.Errorf("unexpected content-type: %s", ct) + return "", fmt.Errorf("unexpected content-type: %s", ct) } - return nil + advertisement, err := io.ReadAll(io.LimitReader(resp.Body, 64*1024)) + if err != nil { + return "", fmt.Errorf("reading upload-pack advertisement: %w", err) + } + if bytes.Contains(advertisement, []byte("object-format=sha256")) { + return models.ObjectFormatSHA256, nil + } + + return models.ObjectFormatSHA1, nil +} + +func (r *Resyncer) suspendUnsupported(ctx context.Context, repo *models.Repo) (bool, error) { + if err := r.gitm.Delete(repo); err != nil { + r.logger.Warn("failed to remove local clone of suspended repo", "did", repo.RepoDid, "err", err) + } + + repo.State = models.RepoStateSuspended + repo.ErrorMsg = "unsupported sha256 object format" + repo.RetryCount = 0 + repo.RetryAfter = 0 + if err := db.UpsertRepo(ctx, r.db, repo); err != nil { + return false, fmt.Errorf("suspending sha256 repo: %w", err) + } + + r.logger.Info("suspended sha256 repo, reads forwarded to knot", "did", repo.RepoDid, "knot", repo.KnotDomain) + return true, nil } func (r *Resyncer) handleResyncFailure(ctx context.Context, repoDid syntax.DID, err error) error { diff --git a/knotmirror/resyncer_test.go b/knotmirror/resyncer_test.go new file mode 100644 --- /dev/null +++ b/knotmirror/resyncer_test.go @@ -0,0 +1,132 @@ +package knotmirror + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "tangled.org/core/knotmirror/config" + "tangled.org/core/knotmirror/models" +) + +func uploadPackAdvert(capabilities string) string { + return "001e# service=git-upload-pack\n" + + "0000" + + "0000000000000000000000000000000000000000 capabilities^{}\x00" + capabilities + "\n" + + "0000" +} + +func TestCheckKnotObjectFormat(t *testing.T) { + const gitContentType = "application/x-git-upload-pack-advertisement" + + tests := []struct { + name string + status int + contentType string + body string + wantFormat models.ObjectFormat + wantErr bool + wantRateLimit bool + }{ + { + name: "sha256 repo is detected", + status: http.StatusOK, + contentType: gitContentType, + body: uploadPackAdvert("multi_ack thin-pack side-band-64k ofs-delta object-format=sha256 agent=git/2.45.0"), + wantFormat: models.ObjectFormatSHA256, + }, + { + name: "explicit sha1 repo stays on sha1", + status: http.StatusOK, + contentType: gitContentType, + body: uploadPackAdvert("multi_ack thin-pack side-band-64k ofs-delta object-format=sha1 agent=git/2.45.0"), + wantFormat: models.ObjectFormatSHA1, + }, + { + name: "advertisement without object-format defaults to sha1", + status: http.StatusOK, + contentType: gitContentType, + body: uploadPackAdvert("multi_ack thin-pack side-band-64k ofs-delta agent=git/2.34.0"), + wantFormat: models.ObjectFormatSHA1, + }, + { + name: "sha256 detected with content-type parameters", + status: http.StatusOK, + contentType: gitContentType + "; charset=utf-8", + body: uploadPackAdvert("object-format=sha256"), + wantFormat: models.ObjectFormatSHA256, + }, + { + name: "rate limited knot", + status: http.StatusTooManyRequests, + wantErr: true, + wantRateLimit: true, + }, + { + name: "missing repo on knot", + status: http.StatusNotFound, + wantErr: true, + }, + { + name: "non git content-type", + status: http.StatusOK, + contentType: "text/html", + body: "not a git server", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var gotPath string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + if tt.contentType != "" { + w.Header().Set("Content-Type", tt.contentType) + } + w.WriteHeader(tt.status) + io.WriteString(w, tt.body) + })) + defer srv.Close() + + r := &Resyncer{ + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + cfg: &config.Config{}, + httpClient: srv.Client(), + } + repo := &models.Repo{ + RepoDid: "did:plc:boltless", + KnotDomain: srv.URL, + } + + format, err := r.checkKnot(context.Background(), repo) + + if tt.wantErr { + if err == nil { + t.Fatalf("expected error, got format %q", format) + } + if format != "" { + t.Errorf("expected empty format on error, got %q", format) + } + if got := isRateLimitError(err); got != tt.wantRateLimit { + t.Errorf("isRateLimitError = %v, want %v", got, tt.wantRateLimit) + } + return + } + + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if format != tt.wantFormat { + t.Errorf("format = %q, want %q", format, tt.wantFormat) + } + if !strings.HasSuffix(gotPath, "/info/refs") { + t.Errorf("expected upstream request to /info/refs, got %q", gotPath) + } + }) + } +} diff --git a/knotmirror/models/models.go b/knotmirror/models/models.go --- a/knotmirror/models/models.go +++ b/knotmirror/models/models.go @@ -56,6 +56,13 @@ return s == RepoStateResyncing } +type ObjectFormat string + +const ( + ObjectFormatSHA1 ObjectFormat = "sha1" + ObjectFormatSHA256 ObjectFormat = "sha256" +) + type HostCursor struct { Hostname string LastSeq int64 diff --git a/knotmirror/xrpc/proxy.go b/knotmirror/xrpc/proxy.go --- a/knotmirror/xrpc/proxy.go +++ b/knotmirror/xrpc/proxy.go @@ -1,6 +1,7 @@ package xrpc import ( + "cmp" "context" "errors" "fmt" @@ -8,10 +9,13 @@ "maps" "net/http" "net/url" + "path" "strings" + "github.com/bluesky-social/indigo/atproto/atclient" "github.com/bluesky-social/indigo/atproto/syntax" indigoxrpc "github.com/bluesky-social/indigo/xrpc" + "github.com/go-git/go-git/v5/plumbing/filemode" "tangled.org/core/api/tangled" "tangled.org/core/knotmirror/db" "tangled.org/core/knotmirror/models" @@ -26,6 +30,7 @@ tangled.GitTempGetTagNSID: tangled.RepoTagNSID, tangled.GitTempGetArchiveNSID: tangled.RepoArchiveNSID, tangled.GitTempListLanguagesNSID: tangled.RepoLanguagesNSID, + tangled.GitTempGetBlobNSID: tangled.RepoBlobNSID, } var hopByHopHeaders = map[string]bool{ @@ -181,4 +186,124 @@ x.logger.Info("proxy: served from knot", "repo", repoDid, "knot", knot.baseURL, "status", resp.StatusCode) return true +} + +func (x *Xrpc) forwardSuspended(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + repoDid, err := syntax.ParseDID(r.URL.Query().Get("repo")) + if err != nil { + next.ServeHTTP(w, r) + return + } + + repo, err := db.GetRepoByRepoDid(r.Context(), x.db, repoDid) + if err != nil || repo == nil || repo.State != models.RepoStateSuspended { + next.ServeHTTP(w, r) + return + } + + nsid := strings.TrimPrefix(r.URL.Path, "/xrpc/") + switch nsid { + case tangled.GitTempGetEntryNSID: + x.serveSuspendedEntry(w, r, repoDid) + case tangled.GitTempGetBlobNSID: + q := r.URL.Query() + q.Set("raw", "true") + r.URL.RawQuery = q.Encode() + x.forwardOrFail(w, r, repoDid) + default: + if _, ok := mirrorToKnotNSID[nsid]; !ok { + next.ServeHTTP(w, r) + return + } + x.forwardOrFail(w, r, repoDid) + } + }) +} + +func (x *Xrpc) forwardOrFail(w http.ResponseWriter, r *http.Request, repoDid syntax.DID) { + if x.proxyToKnot(w, r, repoDid) { + return + } + writeJson(w, http.StatusBadGateway, atclient.ErrorBody{Name: "BadGateway", Message: "failed to reach knot for suspended repo"}) +} + +func (x *Xrpc) serveSuspendedEntry(w http.ResponseWriter, r *http.Request, repoDid syntax.DID) { + ref := cmp.Or(r.URL.Query().Get("ref"), "HEAD") + filePath := r.URL.Query().Get("path") + if filePath == "" { + writeJson(w, http.StatusBadRequest, atclient.ErrorBody{Name: "BadRequest", Message: "missing path parameter"}) + return + } + + knot, err := x.resolveKnot(r.Context(), repoDid) + if err != nil { + x.logger.Warn("suspended entry: failed to resolve knot", "repo", repoDid, "err", err) + writeJson(w, http.StatusBadGateway, atclient.ErrorBody{Name: "BadGateway", Message: "failed to resolve knot for suspended repo"}) + return + } + + client := &indigoxrpc.Client{Host: knot.baseURL, Client: x.httpClient} + out, err := tangled.RepoBlob(r.Context(), client, filePath, false, ref, knot.repoIdentifier) + if err != nil { + x.logger.Warn("suspended entry: knot repo.blob failed", "repo", repoDid, "err", err) + writeJson(w, http.StatusBadGateway, atclient.ErrorBody{Name: "BadGateway", Message: "failed to read entry from knot"}) + return + } + + mode := filemode.Regular + if out.Submodule != nil { + mode = filemode.Submodule + } + + writeJson(w, http.StatusOK, tangled.GitTempGetEntry_Output{ + Name: path.Base(filePath), + Mode: mode.String(), + Size: derefInt64(out.Size), + LastCommit: suspendedLastCommit(out.LastCommit), + Submodule: suspendedSubmodule(out.Submodule), + }) +} + +func suspendedLastCommit(c *tangled.RepoBlob_LastCommit) *tangled.GitTempDefs_Commit { + if c == nil || c.Author == nil { + return nil + } + sig := suspendedSignature(c.Author) + hash := c.Hash + return &tangled.GitTempDefs_Commit{ + Author: sig, + Committer: sig, + Hash: &hash, + Message: c.Message, + } +} + +func suspendedSignature(s *tangled.RepoBlob_Signature) *tangled.GitTempDefs_Signature { + if s == nil { + return nil + } + return &tangled.GitTempDefs_Signature{ + Name: s.Name, + Email: s.Email, + When: s.When, + } +} + +func suspendedSubmodule(s *tangled.RepoBlob_Submodule) *tangled.GitTempDefs_Submodule { + if s == nil { + return nil + } + return &tangled.GitTempDefs_Submodule{ + Name: s.Name, + Url: s.Url, + Branch: s.Branch, + } +} + +func derefInt64(v *int64) int64 { + if v == nil { + return 0 + } + return *v } diff --git a/knotmirror/xrpc/xrpc.go b/knotmirror/xrpc/xrpc.go --- a/knotmirror/xrpc/xrpc.go +++ b/knotmirror/xrpc/xrpc.go @@ -58,6 +58,7 @@ r.Group(func(r chi.Router) { r.Use(x.inflight.middleware) + r.Use(x.forwardSuspended) r.Get("/"+tangled.GitTempGetArchiveNSID, x.GetArchive) r.Get("/"+tangled.GitTempGetBlobNSID, x.GetBlob)