From 090ed7ff8fd4a83fb4fc7ed40892eeb831776d40 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Tue, 9 Jun 2026 14:09:05 +0900 Subject: [PATCH] knotmirror: request index on refUpdate THIS IS BAD DESIGN. A. Indexer should directly subscribe to Knot event stream. Push-to-index is only for admin use and some edge cases. B. Index should be triggered just after the refUpdate event and not after the full git mirror resync is done. We are doing this because we don't have unified knotstream with basic repository state (ref list) passed as event payload. Should be updated later. Signed-off-by: Seongmin Lee --- knotmirror/config/config.go | 5 ++++ knotmirror/git.go | 48 +++++++++++++++++++++++++++++++++++++ knotmirror/resyncer.go | 48 +++++++++++++++++++++++++++++++++++++ 3 files changed, 101 insertions(+) diff --git a/knotmirror/config/config.go b/knotmirror/config/config.go index 51173b1d..a3097b47 100644 --- a/knotmirror/config/config.go +++ b/knotmirror/config/config.go @@ -16,6 +16,7 @@ type Config struct { KnotSSRF bool `env:"MIRROR_KNOT_SSRF, default=false"` GitRepoBasePath string `env:"MIRROR_GIT_BASEPATH, default=repos"` GitRepoFetchTimeout time.Duration `env:"MIRROR_GIT_FETCH_TIMEOUT, default=600s"` + Search SearchConfig `env:",prefix=MIRROR_SEARCH_"` ResyncParallelism int `env:"MIRROR_RESYNC_PARALLELISM, default=5"` Slurper SlurperConfig `env:",prefix=MIRROR_SLURPER_"` UseSSL bool `env:"MIRROR_USE_SSL, default=false"` @@ -37,6 +38,10 @@ type SlurperConfig struct { ConcurrencyPerHost int `env:"CONCURRENCY, default=4"` } +type SearchConfig struct { + ZoektUrl string `env:"ZOEKT_URL"` // base url to zoekt node. skipped when empty +} + func Load(ctx context.Context) (*Config, error) { var cfg Config if err := envconfig.Process(ctx, &cfg); err != nil { diff --git a/knotmirror/git.go b/knotmirror/git.go index 79d3776e..7a0238d5 100644 --- a/knotmirror/git.go +++ b/knotmirror/git.go @@ -17,6 +17,11 @@ import ( "tangled.org/core/knotmirror/models" ) +type branch struct { + Name string `json:"name"` + Version string `json:"version"` +} + type GitMirrorManager interface { Exist(repo *models.Repo) (bool, error) // Clone clones the repository as a mirror @@ -25,6 +30,7 @@ type GitMirrorManager interface { Fetch(ctx context.Context, repo *models.Repo) error // Sync mirrors the repository. It will clone the repository if repository doesn't exist. Sync(ctx context.Context, repo *models.Repo) error + DefaultBranch(ctx context.Context, repo *models.Repo) (branch, error) Delete(repo *models.Repo) error } @@ -149,6 +155,33 @@ func (c *CliGitMirrorManager) Sync(ctx context.Context, repo *models.Repo) error return nil } +func (c *CliGitMirrorManager) DefaultBranch(ctx context.Context, repo *models.Repo) (branch, error) { + path := c.makeRepoPath(repo) + + nameCmd := exec.CommandContext(ctx, "git", "-C", path, "symbolic-ref", "--short", "HEAD") + nameOut, err := nameCmd.Output() + if err != nil { + return branch{}, err + } + + // --verify --quiet exits 1 with no output on an empty repo (unborn HEAD). + revCmd := exec.CommandContext(ctx, "git", "-C", path, "rev-parse", "--verify", "--quiet", "HEAD") + revOut, err := revCmd.Output() + if err != nil { + return branch{}, err + } + + version := strings.TrimSpace(string(revOut)) + if version == "" { + return branch{}, errors.New("git: no commits") + } + + return branch{ + Name: strings.TrimSpace(string(nameOut)), + Version: version, + }, nil +} + func (c *CliGitMirrorManager) Delete(repo *models.Repo) error { return os.RemoveAll(c.makeRepoPath(repo)) } @@ -288,6 +321,21 @@ func (c *GoGitMirrorManager) Sync(ctx context.Context, repo *models.Repo) error return nil } +func (c *GoGitMirrorManager) DefaultBranch(ctx context.Context, repo *models.Repo) (branch, error) { + gr, err := git.PlainOpen(c.makeRepoPath(repo)) + if err != nil { + return branch{}, fmt.Errorf("opening local repo: %w", err) + } + ref, err := gr.Head() + if err != nil { + return branch{}, fmt.Errorf("resolving HEAD: %w", err) + } + return branch{ + Name: ref.Name().Short(), + Version: ref.Hash().String(), + }, nil +} + func (c *GoGitMirrorManager) Delete(repo *models.Repo) error { return os.RemoveAll(c.makeRepoPath(repo)) } diff --git a/knotmirror/resyncer.go b/knotmirror/resyncer.go index 9095e103..3072651b 100644 --- a/knotmirror/resyncer.go +++ b/knotmirror/resyncer.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "database/sql" + "encoding/json" "errors" "fmt" "io" @@ -258,6 +259,24 @@ func (r *Resyncer) doResync(ctx context.Context, repoDid syntax.DID) (bool, erro return false, err } + // request index to zoekt server + // NOTE: indexing after full git resync is bad design. We are doing _after_ the sync because knotstream event doesn't include repository refs. + // NOTE: and zoekt indexer should directly subscribe to the knot. remove this when we have knotrelay. + if r.cfg.Search.ZoektUrl != "" { + go func() { + idxCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + defaultBranch, err := r.gitm.DefaultBranch(idxCtx, repo) + if err != nil { + r.logger.Warn("resolving default branch for indexing failed", "did", repo.RepoDid, "error", err) + return + } + if err := r.requestIndex(idxCtx, repo.RepoDid, []branch{defaultBranch}); err != nil { + r.logger.Warn("requesting zoekt index failed", "did", repo.RepoDid, "err", err) + } + }() + } + // queue repo_stats_update job r.indexer.AddTask(context.TODO(), &knotstream.Task{Key: repo.RepoDid.String()}) @@ -396,3 +415,32 @@ func backoff(retries int, max int) time.Duration { jitter := time.Millisecond * time.Duration(rand.Intn(1000)) return time.Second*time.Duration(dur) + jitter } + +func (r *Resyncer) requestIndex(ctx context.Context, repoDid syntax.DID, branches []branch) error { + r.logger.Info("requesting index", "repo", repoDid, "branches", branches) + body, err := json.Marshal(map[string]any{ + "repo": repoDid.String(), + "branches": branches, + }) + if err != nil { + return fmt.Errorf("marshaling index request: %w", err) + } + + endpoint := r.cfg.Search.ZoektUrl + "/indexserver/admin/enqueueIndex" + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := r.httpClient.Do(req) + if err != nil { + return fmt.Errorf("requesting zoekt index: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("non-ok status: %d", resp.StatusCode) + } + return nil +} -- 2.51.2