353 lines
15 KiB
Go
353 lines
15 KiB
Go
package index
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// ErrConflict is what a Store says when a write found the key changed or created since it was read.
|
|
var ErrConflict = errors.New("the key changed since it was read")
|
|
|
|
// ErrUnreadable wraps what says a message cannot be indexed, ever: it is never tried again.
|
|
var ErrUnreadable = errors.New("unreadable message")
|
|
|
|
// Store is the candidates bucket: a compare-and-set key-value store.
|
|
type Store interface {
|
|
// Get is the key's value and revision; found is false when there is none.
|
|
Get(ctx context.Context, key string) (value []byte, revision uint64, found bool, err error)
|
|
// Create writes a key that must not exist (ErrConflict when it does).
|
|
Create(ctx context.Context, key string, value []byte) error
|
|
// Update writes the key if its revision is still revision (ErrConflict when not).
|
|
Update(ctx context.Context, key string, value []byte, revision uint64) error
|
|
// Delete removes the key if its revision is still revision (ErrConflict when not).
|
|
Delete(ctx context.Context, key string, revision uint64) error
|
|
// Keys is every key the store holds now; none is an empty list, not an error.
|
|
Keys(ctx context.Context) ([]string, error)
|
|
}
|
|
|
|
// trunkLike are the branches that are no group: a pull request from one is nobody's delivery group (the rule of
|
|
// mesh-catalog modules/mesh-delivery/cmd/mesh-delivery/holder.go:321).
|
|
var trunkLike = map[string]bool{"main": true, "master": true, "trunk": true, "": true}
|
|
|
|
// writeTries bounds the reads again after a conflict; a message that still conflicts is left to the server to deliver again.
|
|
const writeTries = 8
|
|
|
|
// Indexer writes candidates.<group> from the forge bridge's messages.
|
|
type Indexer struct {
|
|
Store Store // candidates
|
|
// Modules is the modules bucket; nil writes no fingerprints.
|
|
Modules Store
|
|
// Manifests is the manifests bucket; nil writes no manifests (issue 1449).
|
|
Manifests Store
|
|
// Taken is the taken bucket: per repository the last merge taken in (taken.go, ADR 0556); nil writes none.
|
|
Taken Store
|
|
// Now is the clock of a taken value; nil is time.Now.
|
|
Now func() time.Time
|
|
Builds *Builds
|
|
// Named asks the forge for the git object of each named path of the module directories given, at a commit
|
|
// (gitea.gitea_commit_trees with paths; novox/hq ADR 0499 clauses 2 and 3): the indexer is the one reader of a
|
|
// manifest's paths (ADR 0495 clause 1, the lead's decision L8 of R4e). Nil asks nothing, and a module naming paths
|
|
// has none.
|
|
Named func(ctx context.Context, repository, sha string, paths map[string][]string) (map[string]map[string]string, error)
|
|
Trunk func(ctx context.Context, repository, sha string) ([]string, error)
|
|
// Read asks the forge for a repository's trees at a commit and the manifest and lock-file hashes of the one module directory dir (the
|
|
// root spelled ""), as the forge bridge reads them for a merged message (registered.go). Nil indexes no registered module.
|
|
Read func(ctx context.Context, repository, sha, dir string) (RepoRead, error)
|
|
// Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains).
|
|
Contains func(ctx context.Context, repository, commit, in string) (bool, error)
|
|
Logf func(format string, args ...any)
|
|
// Placement is the placement bucket (delivery.PlacementBucket, ADR 0549 §2); nil writes no placement.
|
|
Placement Store
|
|
// PlacementCandidates is the placement candidates bucket (delivery.PlacementCandidatesBucket, issue 1126); nil writes
|
|
// no candidate (placement_candidates.go).
|
|
PlacementCandidates Store
|
|
// ReadPlacement asks the forge for the catalogue's placement now, at the trunk's head and each open pull request's
|
|
// (indexer_placement_reread, ADR 0559 §8); nil rereads none.
|
|
ReadPlacement func(ctx context.Context) (PlacementRead, error)
|
|
// Say says one of the indexer's events (placement-changed); nil says none.
|
|
Say func(event string, body any) error
|
|
// Drained says whether the indexer has applied every message of the forge stream (the consumer has none pending and none
|
|
// unacknowledged); FillTaken writes the source's heads only then. Nil fills none from the source (taken.go).
|
|
Drained func(ctx context.Context) (bool, error)
|
|
|
|
mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
|
|
source sourceNow
|
|
previews map[string]delivery.Preview // by previews key (previews.go)
|
|
toolchains map[string]map[string]delivery.Toolchain // language, then node (toolchains.go)
|
|
tried map[string]bool // repository, commit and directory a registered module's manifest was tried at (registered.go)
|
|
unread map[string]bool // registered modules whose last read the forge could not answer, read again by RetryUnread
|
|
retrying atomic.Bool // a RetryUnread runs in the background (kickRetry)
|
|
retryAgain atomic.Bool // an event asked for one more RetryUnread after the running one
|
|
retryWG sync.WaitGroup // the background RetryUnread, for tests to wait on
|
|
recompute sync.Mutex // one Recompute at a time
|
|
untaken map[string]string // repository to the latest merge applied while the taken state was missing (taken.go)
|
|
filling atomic.Bool // a FillTaken runs in the background (KickFill)
|
|
fillAgain atomic.Bool // a kick arrived while it ran
|
|
}
|
|
|
|
func (ix *Indexer) logf(format string, args ...any) {
|
|
if ix.Logf != nil {
|
|
ix.Logf(format, args...)
|
|
}
|
|
}
|
|
|
|
// Apply takes one message of the bridge's stream FORGE into the group of its branch. A pull message adds or moves the
|
|
// pull request, a closed one or a merged message takes it out; a group left with none is deleted. A change is written
|
|
// by compare-and-set on the revision read, and read again on a conflict. A lone pull request is a group of one. A
|
|
// merged message also writes the fingerprint of each module of its repository (modules.go).
|
|
func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error {
|
|
if f.Kind == delivery.ForgeOpen {
|
|
return ix.applyOpen(ctx, f)
|
|
}
|
|
if f.Kind != delivery.ForgePull && f.Kind != delivery.ForgeMerged {
|
|
return nil
|
|
}
|
|
if f.Kind == delivery.ForgeMerged && ix.Modules != nil {
|
|
if err := ix.applyMerged(ctx, f); err != nil {
|
|
return err
|
|
}
|
|
if err := ix.writeTaken(ctx, f); err != nil {
|
|
return err
|
|
}
|
|
ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945)
|
|
}
|
|
if err := ix.applyPlacement(ctx, f); err != nil {
|
|
return err
|
|
}
|
|
if err := ix.applyPlacementCandidates(ctx, f); err != nil {
|
|
return err
|
|
}
|
|
var digests map[string]string
|
|
var held map[string]delivery.Module
|
|
if f.Kind == delivery.ForgePull {
|
|
var err error
|
|
if digests, err = ix.keepManifests(ctx, f); err != nil {
|
|
return err
|
|
}
|
|
if held, err = ix.heldModules(ctx); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if trunkLike[f.Branch] {
|
|
ix.logf("%s#%d: its branch %q is no delivery group, so no candidate", f.Repository, f.Number, f.Branch)
|
|
return nil
|
|
}
|
|
key, err := delivery.CandidatesKey(f.Branch)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err)
|
|
}
|
|
return write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
|
|
var cur *delivery.Candidate
|
|
if found {
|
|
c, err := delivery.DecodeCandidate(value)
|
|
if err != nil {
|
|
return nil, false, fmt.Errorf("candidates.%s holds what is not a candidate: %w", key, err)
|
|
}
|
|
cur = &c
|
|
}
|
|
next := ix.next(cur, f, digests, held)
|
|
if next != nil {
|
|
ix.withPreviews(ctx, next)
|
|
if err := ix.refuseDisagreement(*next); err != nil {
|
|
return nil, false, err
|
|
}
|
|
}
|
|
switch {
|
|
case next == nil && cur == nil, next != nil && cur != nil && delivery.SameCandidate(*cur, *next):
|
|
return nil, false, nil
|
|
case next == nil:
|
|
return nil, true, nil
|
|
}
|
|
data, err := delivery.EncodeCandidate(*next)
|
|
return data, false, err
|
|
})
|
|
}
|
|
|
|
// applyOpen takes an open message into the candidates (novox/hq ADR 0443 §5): every pull request it does not name is
|
|
// taken out of its group, and a group left with none is deleted. What the message names is left as it is. A pull request
|
|
// opened after the bridge read the forge is heard after this message and added again by its own.
|
|
func (ix *Indexer) applyOpen(ctx context.Context, f delivery.Forge) error {
|
|
named := make(map[delivery.OpenPull]bool, len(f.Open))
|
|
for _, p := range f.Open {
|
|
named[p] = true
|
|
}
|
|
if ix.PlacementCandidates != nil {
|
|
// A pull request the forge no longer has open keeps no placement candidate (issue 1126).
|
|
if err := ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool {
|
|
return !named[delivery.OpenPull{Repository: c.Pull.Repository, Number: c.Pull.Number}]
|
|
}, "its pull request is not open"); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
keys, err := ix.Store.Keys(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("listing the candidates: %w", err)
|
|
}
|
|
for _, key := range keys {
|
|
err := write(ctx, ix.Store, key, func(value []byte, found bool) ([]byte, bool, error) {
|
|
if !found {
|
|
return nil, false, nil
|
|
}
|
|
c, err := delivery.DecodeCandidate(value)
|
|
if err != nil {
|
|
return nil, false, fmt.Errorf("candidates.%s holds what is not a candidate: %w", key, err)
|
|
}
|
|
kept := slices.DeleteFunc(slices.Clone(c.Pulls), func(p delivery.CandidatePull) bool {
|
|
return !named[delivery.OpenPull{Repository: p.Repository, Number: p.Number}]
|
|
})
|
|
switch {
|
|
case len(kept) == len(c.Pulls):
|
|
return nil, false, nil
|
|
case len(kept) == 0:
|
|
ix.logf("%s: no pull request of the group is open, so no candidate", key)
|
|
return nil, true, nil
|
|
}
|
|
heads := make([]delivery.Head, len(kept))
|
|
for i, p := range kept {
|
|
heads[i] = p.Head
|
|
}
|
|
hash, err := delivery.HeadsHash(heads)
|
|
if err != nil {
|
|
return nil, false, fmt.Errorf("%w: candidates.%s: %v", ErrUnreadable, key, err)
|
|
}
|
|
c.Pulls, c.HeadsHash = kept, hash
|
|
data, err := delivery.EncodeCandidate(c)
|
|
return data, false, err
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// write changes one key of a store by compare-and-set: compute is given the value read (found says there is one) and
|
|
// answers the new value, or del to remove the key, or neither for no change. A conflict reads again, up to writeTries.
|
|
func write(ctx context.Context, st Store, key string, compute func(value []byte, found bool) (next []byte, del bool, err error)) error {
|
|
for try := 0; try < writeTries; try++ {
|
|
value, revision, found, err := st.Get(ctx, key)
|
|
if err != nil {
|
|
return fmt.Errorf("reading %s: %w", key, err)
|
|
}
|
|
next, del, err := compute(value, found)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
switch {
|
|
case next == nil && !del:
|
|
return nil
|
|
case del:
|
|
err = st.Delete(ctx, key, revision)
|
|
case !found:
|
|
err = st.Create(ctx, key, next)
|
|
default:
|
|
err = st.Update(ctx, key, next, revision)
|
|
}
|
|
if errors.Is(err, ErrConflict) {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("writing %s: %w", key, err)
|
|
}
|
|
return nil
|
|
}
|
|
return fmt.Errorf("%s changed under every one of %d tries", key, writeTries)
|
|
}
|
|
|
|
// next is the group after the message; nil when the group is left with no pull request.
|
|
//
|
|
// A pull message that carries manifests (issue 1448) gives each module it touches, in the pull request's Fps, the digest of
|
|
// its manifest and why it has no fingerprint of the new form (NoneNoPreview), and the candidate its FPSet (issue 1451); a
|
|
// pull request whose head is held again without its directories keeps the Fps it had. digests are the manifests' digests by
|
|
// directory; held are the modules the modules bucket holds, which name the module of a changed directory (moduleOfDir).
|
|
func (ix *Indexer) next(cur *delivery.Candidate, f delivery.Forge, digests map[string]string, held map[string]delivery.Module) *delivery.Candidate {
|
|
var pulls []delivery.CandidatePull
|
|
var base uint64
|
|
if cur != nil {
|
|
pulls, base = slices.Clone(cur.Pulls), cur.Base
|
|
}
|
|
at := slices.IndexFunc(pulls, func(p delivery.CandidatePull) bool { return p.Repository == f.Repository && p.Number == f.Number })
|
|
if f.Kind == delivery.ForgeMerged || f.Action == delivery.PullClosed {
|
|
if at < 0 {
|
|
return cur
|
|
}
|
|
pulls = slices.Delete(pulls, at, at+1)
|
|
} else {
|
|
entry := delivery.CandidatePull{Head: delivery.Head{Repository: f.Repository, Number: f.Number, Head: f.Head}, Modules: []string{}}
|
|
switch {
|
|
case f.DirsKnown:
|
|
for _, dir := range f.Dirs {
|
|
m := moduleOfDir(held, f, dir)
|
|
if !slices.Contains(entry.Modules, m) {
|
|
entry.Modules = append(entry.Modules, m)
|
|
}
|
|
if digest, ok := digests[dir]; ok {
|
|
if entry.Fps == nil {
|
|
entry.Fps = map[string]delivery.CandidateModule{}
|
|
}
|
|
entry.Fps[m] = delivery.CandidateModule{Manifest: digest, None2: NoneNoPreview}
|
|
}
|
|
}
|
|
case at >= 0 && pulls[at].Head.Head == f.Head:
|
|
// Its directories are not known now, for the head the group already holds: what it knew stands.
|
|
entry.Modules, entry.Fps = pulls[at].Modules, pulls[at].Fps
|
|
}
|
|
if at >= 0 {
|
|
pulls[at] = entry
|
|
} else {
|
|
pulls = append(pulls, entry)
|
|
}
|
|
}
|
|
if len(pulls) == 0 {
|
|
return nil
|
|
}
|
|
slices.SortFunc(pulls, func(a, b delivery.CandidatePull) int {
|
|
if c := strings.Compare(a.Repository, b.Repository); c != 0 {
|
|
return c
|
|
}
|
|
return a.Number - b.Number
|
|
})
|
|
heads := make([]delivery.Head, len(pulls))
|
|
for i, p := range pulls {
|
|
heads[i] = p.Head
|
|
}
|
|
hash, _ := delivery.HeadsHash(heads)
|
|
return &delivery.Candidate{V: delivery.Version, Group: f.Branch, Base: base, HeadsHash: hash, Pulls: pulls, FPSet: delivery.FingerprintSet(pulls)}
|
|
}
|
|
|
|
// NoneNoPreview is why a candidate's touched module has no fingerprint of the new form: the merge preview is not
|
|
// computed in this release (novox/hq ADR 0465 clause 3; the checker comes with issue 1445).
|
|
const NoneNoPreview = "no merge preview is computed"
|
|
|
|
// refuseDisagreement refuses a candidate in which two pull requests give one module different manifests, as the
|
|
// candidate's own check does (mesh-sdk Candidate.check), naming both pull requests so the log says which; nothing is
|
|
// written, and the message is dropped as unreadable and said in the log (main.go handle): the next message of either pull
|
|
// request is judged again.
|
|
func (ix *Indexer) refuseDisagreement(c delivery.Candidate) error {
|
|
first := map[string]delivery.CandidatePull{}
|
|
for _, p := range c.Pulls {
|
|
for module, m := range p.Fps {
|
|
before, ok := first[module]
|
|
if !ok {
|
|
first[module] = p
|
|
continue
|
|
}
|
|
if !reflect.DeepEqual(before.Fps[module], m) {
|
|
err := fmt.Errorf("%w: candidates.%s: %s#%d and %s#%d give the module %q different manifests (%s and %s), so the group is not written",
|
|
ErrUnreadable, c.Group, before.Repository, before.Number, p.Repository, p.Number, module, before.Fps[module].Manifest, m.Manifest)
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|