package sync
import (
"context"
"errors"
"fmt"
"os"
"sort"
"strings"
"github.com/danmestas/go-libfossil/internal/content"
"github.com/danmestas/go-libfossil/internal/manifest"
"github.com/danmestas/go-libfossil/internal/repo"
"github.com/danmestas/go-libfossil/internal/shun"
"github.com/danmestas/go-libfossil/internal/xfer"
"github.com/danmestas/go-libfossil/simio"
)
// phantomStallSampleSize caps how many missing UUIDs PhantomStallError
// carries, so a repository with thousands of phantoms doesn't produce an
// unreadable error message.
const phantomStallSampleSize = 10
// processResponseCancelCheckStride is how many cards processResponse handles
// between context-cancellation checks. One clone batch is processed card by
// card with no network wait in between, so without an inner check a cancelled
// context would not be observed until the next round begins; the check is
// batched to avoid a channel read per card.
const processResponseCancelCheckStride = 256
// cloneAvailableDeltaSourceLimit bounds the positive-only availability memo
// used while processing received delta cards. Once full, correctness falls
// back to querying the repository for subsequent sources.
const cloneAvailableDeltaSourceLimit = 4096
// PhantomStallError reports a clone that stopped making progress while
// phantom blobs — content referenced but never received — were still
// outstanding. Terminating a stalled clone is correct; reporting it as
// success is not, so this is always returned instead of nil in that case.
// Count is programmatically reachable so a caller can act on it without
// parsing the message.
type PhantomStallError struct {
Count int // phantom UUIDs still outstanding when the clone stopped
Sample []string // up to phantomStallSampleSize lexicographically-smallest of those UUIDs, for diagnosis
}
func (e *PhantomStallError) Error() string {
return fmt.Sprintf("sync.Clone: stalled with %d phantom artifact(s) outstanding, e.g. %s",
e.Count, strings.Join(e.Sample, ", "))
}
// newPhantomStallError builds a PhantomStallError from the session's current
// phantom set. Called only once that set is known to be non-empty.
func newPhantomStallError(phantoms map[string]bool) *PhantomStallError {
if len(phantoms) == 0 {
panic("sync.newPhantomStallError: phantoms must not be empty")
}
all := make([]string, 0, len(phantoms))
for uuid := range phantoms {
all = append(all, uuid)
}
sort.Strings(all)
if len(all) > phantomStallSampleSize {
all = all[:phantomStallSampleSize]
}
return &PhantomStallError{Count: len(phantoms), Sample: all}
}
// cloneSession holds the mutable state of a running clone.
type cloneSession struct {
repo *repo.Repo
env *simio.Env
opts CloneOpts
result CloneResult
linker *manifest.ReceiveLinker
phantoms map[string]bool
availableDeltaSources map[string]struct{}
seqno int
projectCode string
serverCode string
roundRecvd int // file/cfile cards the last reply delivered
roundPush bool // the last reply carried a push card
cleanup bool // the next request is an operation-less cleanup
serverErr string // a server error after the first round, which fails the clone
cookie string // server-issued snapshot bound; echoed every round so the
// server can scope its blob query to the rid range that existed when the
// clone session opened. See cloneCapPrefix in handler.go.
obs Observer
}
// Clone performs a full repository clone from a remote Fossil server.
// It creates a new repository at path, runs the clone protocol until
// convergence, and returns the opened repo and a result summary.
// On error, the partially-created repo file is removed.
func Clone(ctx context.Context, path string, t Transport, opts CloneOpts) (r *repo.Repo, result *CloneResult, err error) {
if path == "" {
panic("sync.Clone: path must not be empty")
}
if t == nil {
panic("sync.Clone: t must not be nil")
}
env, storage := cloneEnv(opts)
// Path must not already exist.
if _, statErr := storage.Stat(path); statErr == nil {
return nil, nil, fmt.Errorf("sync.Clone: file already exists: %s", path)
}
user := opts.User
if user == "" {
user = "setup"
}
// Create the repository.
r, err = repo.CreateWithEnv(path, user, env, "")
if err != nil {
return nil, nil, fmt.Errorf("sync.Clone: create repo: %w", err)
}
// Cleanup on error: close repo and remove the file.
defer func() {
if err != nil {
r.Close()
if rmErr := storage.Remove(path); rmErr != nil {
fmt.Fprintf(os.Stderr, "sync.Clone: cleanup failed: %v\n", rmErr)
}
r = nil
}
}()
if err = clearInheritedConfig(r); err != nil {
return
}
linker, linkerErr := manifest.NewReceiveLinker(r)
if linkerErr != nil {
err = fmt.Errorf("sync.Clone: create receive linker: %w", linkerErr)
return
}
cs := &cloneSession{
repo: r,
env: env,
opts: opts,
linker: linker,
phantoms: make(map[string]bool),
availableDeltaSources: make(map[string]struct{}),
seqno: 1,
}
cs.obs = resolveObserver(opts.Observer)
result, err = cs.complete(ctx, t)
if err != nil {
return
}
return r, result, nil
}
// cloneEnv is the environment a clone runs in and the storage its file
// lives in.
func cloneEnv(opts CloneOpts) (*simio.Env, simio.Storage) {
env := opts.Env
if env == nil {
env = simio.RealEnv()
}
storage := env.Storage
if storage == nil {
storage = simio.OSStorage{}
}
return env, storage
}
// complete runs the clone session, links what it received and fetches any
// artifact it left missing. The result is meaningful on error too: callers
// inspecting it (e.g. BlobsRecvd) alongside a *PhantomStallError shouldn't
// see it dropped to nil.
func (cs *cloneSession) complete(ctx context.Context, t Transport) (*CloneResult, error) {
cloneResult, err := cs.run(ctx, t)
if err != nil {
return cloneResult, err
}
// The receive linker has already handled artifacts as their verified
// content arrived. Finalize retries deferred candidates and runs the
// bounded residual sweep with that same cache.
_, finalizeErr := cs.linker.Finalize(ctx)
cloneResult.ArtifactsLinked = cs.linker.Stats().Linked
if finalizeErr != nil {
return cloneResult, fmt.Errorf(
"sync.Clone: finalize receive-linker crosslink: %w", finalizeErr)
}
return cloneResult, cs.fillPhantoms(ctx, t)
}
// clearInheritedConfig drops the config rows a fresh repository is seeded
// with that a clone must take from its source instead.
func clearInheritedConfig(r *repo.Repo) error {
if r == nil {
panic("sync.clearInheritedConfig: nil repo")
}
// Clear project-code — the server will provide its own.
if _, execErr := r.DB().Exec("DELETE FROM config WHERE name='project-code'"); execErr != nil {
return fmt.Errorf("sync.Clone: clear project-code: %w", execErr)
}
// Clear hash-policy for the same reason: a clone adopts its source's
// policy, not the sha3 a fresh repo is seeded with, so a clone of a SHA1
// repo must not start naming artifacts SHA3. The protocol carries no
// config, so dropping the row is what conveys it — an absent row reads
// as "auto", which derives the algorithm from the artifacts that arrive.
if _, execErr := r.DB().Exec("DELETE FROM config WHERE name='hash-policy'"); execErr != nil {
return fmt.Errorf("sync.Clone: clear hash-policy: %w", execErr)
}
return nil
}
// fillPhantoms fetches artifacts the clone left missing (phantoms) with an
// ordinary pull session. The clone session cannot ask for them: clone_seqno
// 0 bars pull from it, and a gimme without pull fetches nothing (draft
// §8.2). A clone of a consistent repository leaves none, and then no pull
// is made. Any still missing after the pull end the clone with a
// *PhantomStallError, as an incomplete repository is not a clone.
func (cs *cloneSession) fillPhantoms(ctx context.Context, t Transport) error {
missing, err := phantomUUIDs(cs.repo)
if err != nil {
return err
}
if len(missing) == 0 {
return nil
}
res, err := Sync(ctx, cs.repo, t, SyncOpts{
Pull: true,
ProjectCode: cs.projectCode,
ServerCode: cs.serverCode,
User: cs.opts.User,
Password: cs.opts.Password,
Env: cs.env,
Buggify: cs.opts.Buggify,
Observer: cs.opts.Observer,
})
if res != nil {
cs.result.Rounds += res.Rounds
cs.result.BlobsRecvd += res.FilesRecvd
}
if err != nil {
return fmt.Errorf("sync.Clone: pull missing artifacts: %w", err)
}
if missing, err = phantomUUIDs(cs.repo); err != nil {
return err
}
if len(missing) > 0 {
return newPhantomStallError(missing)
}
return nil
}
// phantomUUIDs returns the artifacts the repository knows of but lacks.
func phantomUUIDs(r *repo.Repo) (map[string]bool, error) {
rows, err := r.DB().Query(
"SELECT b.uuid FROM phantom p JOIN blob b ON p.rid = b.rid WHERE b.size < 0")
if err != nil {
return nil, fmt.Errorf("sync.Clone: query phantoms: %w", err)
}
missing := make(map[string]bool)
for rows.Next() {
var uuid string
if err := rows.Scan(&uuid); err != nil {
return nil, errors.Join(fmt.Errorf("sync.Clone: scan phantom: %w", err), rows.Close())
}
missing[uuid] = true
}
if err := errors.Join(rows.Err(), rows.Close()); err != nil {
return nil, fmt.Errorf("sync.Clone: read phantoms: %w", err)
}
return missing, nil
}
// run executes the clone loop.
func (cs *cloneSession) run(ctx context.Context, t Transport) (*CloneResult, error) {
ctx = cs.obs.Started(ctx, SessionStart{
Operation: "clone",
Pull: true,
})
for cycle := 0; ; cycle++ {
select {
case <-ctx.Done():
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), ctx.Err())
return &cs.result, ctx.Err()
default:
}
if cycle >= MaxRounds {
err := cs.roundCapError()
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
return &cs.result, err
}
roundCtx := cs.obs.RoundStarted(ctx, cycle)
req, err := cs.buildRequest(cycle)
if err != nil {
cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{})
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
return &cs.result, fmt.Errorf("sync.Clone: buildRequest round %d: %w", cycle, err)
}
resp, err := exchangeWithRetry(ctx, t, req, cs.env.Clock, cs.obs)
if err != nil {
cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{})
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
return &cs.result, fmt.Errorf("sync.Clone: exchange round %d: %w", cycle, err)
}
recvdBefore := cs.result.BlobsRecvd
err = cs.processResponse(ctx, cycle, resp)
if err != nil {
cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{FilesReceived: cs.result.BlobsRecvd - recvdBefore})
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
return &cs.result, fmt.Errorf("sync.Clone: process round %d: %w", cycle, err)
}
cs.result.Rounds = cycle + 1
cs.obs.RoundCompleted(roundCtx, cycle, RoundStats{FilesReceived: cs.result.BlobsRecvd - recvdBefore})
next, nextErr := cs.continueAfter(cycle)
if cs.serverErr != "" && (nextErr != nil || !next) {
nextErr = fmt.Errorf("sync.Clone: server error: %s", cs.serverErr)
}
if nextErr != nil {
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), nextErr)
return &cs.result, nextErr
}
if !next {
break
}
}
if cs.projectCode == "" {
err := fmt.Errorf("sync.Clone: server sent no project code")
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), err)
return &cs.result, err
}
cs.result.ProjectCode = cs.projectCode
cs.result.ServerCode = cs.serverCode
cs.obs.Completed(ctx, sessionEndFromClone(&cs.result), nil)
return &cs.result, nil
}
// roundCapError is the error a clone that ran MaxRounds rounds fails with:
// the server's own error when it sent one, since a server that errs every
// round is the cause, not the round cap.
func (cs *cloneSession) roundCapError() error {
if cs.serverErr != "" {
return fmt.Errorf("sync.Clone: server error: %s", cs.serverErr)
}
return fmt.Errorf("sync.Clone: exceeded %d rounds", MaxRounds)
}
// continueAfter decides, after a reply is processed, whether the clone
// session sends another request and of which kind: draft §3.7, which for a
// clone continues when the round was the first, or stored content (limbs (a)
// and (e); this client sends no private or unversioned cards in a clone, so
// (b) through (d) never hold). The next request reissues `clone 3 NEXT` only
// when the reply carried a push card and the recorded sequence is positive;
// otherwise it is the operation-less cleanup of §8.2, so nothing follows
// clone_seqno 0 but at most one cleanup.
//
// A session that stops while the sequence is still positive did not get the
// whole repository; the draft simply stops, but Clone reports it.
func (cs *cloneSession) continueAfter(cycle int) (bool, error) {
if cycle < 0 {
panic("cloneSession.continueAfter: negative cycle")
}
if cycle > 0 && cs.roundRecvd == 0 {
if cs.seqno > 0 {
return false, fmt.Errorf(
"sync.Clone: server stopped sending at clone sequence %d", cs.seqno)
}
return false, nil
}
cs.cleanup = !cs.roundPush || cs.seqno <= 0
return true, nil
}
// sessionEndFromClone builds a SessionEnd from a CloneResult.
func sessionEndFromClone(r *CloneResult) SessionEnd {
return SessionEnd{
Operation: "clone",
Rounds: r.Rounds,
FilesRecvd: r.BlobsRecvd,
ProjectCode: r.ProjectCode,
Errors: r.Messages,
}
}
// buildRequest assembles one outbound xfer message for a clone round.
func (cs *cloneSession) buildRequest(cycle int) (*xfer.Message, error) {
var cards []xfer.Card
// A clone sends no pragma send-private (CloneOpts has no private mode).
// Were one added, fossil puts it before client-version in cycle 0
// (src/xfer.c client_sync), as headerCards does for a sync (#351).
// Pragma: client-version (every round)
cards = append(cards, versionPragma("client-version"))
// Cookie carries the server-captured clone snapshot bound. Echoing it back
// every round lets the stateless handler enforce a stable upper rid bound
// on its batch query, so a hub that is being written to during a clone
// session can't keep extending the queue and prevent convergence.
if cs.cookie != "" {
cards = append(cards, &xfer.CookieCard{Value: cs.cookie})
}
// The clone card carries the pagination cursor itself — `clone VERSION
// SEQNO`, canonical xfer.c:1553. A clone_seqno card must never go out
// from here: it is server-to-client only, canonical's page_xfer() has no
// parser for it, and sending one lands in the server's unknown-card
// branch as `bad command: clone_seqno N` (issue #74). A cleanup request
// carries no operation card at all (draft §8.2).
if !cs.cleanup {
if cs.seqno <= 0 {
panic("cloneSession.buildRequest: clone reissue after clone_seqno 0")
}
version := cs.opts.Version
if version <= 0 {
version = 3
}
cards = append(cards, &xfer.CloneCard{
Version: version,
SeqNo: cs.seqno,
SeqNoIsDecimal: true,
})
}
// Login card: skip round 0. On round 1+, only if User is set AND projectCode received.
if cycle > 0 && cs.opts.User != "" && cs.projectCode != "" {
loginCard, signed, err := cs.buildLoginCard(cards)
if err != nil {
return nil, fmt.Errorf("clone buildLoginCard: %w", err)
}
cards = append([]xfer.Card{loginCard}, signed...)
} else {
// Every request ends with randomness (draft §3); a login card's
// signature already covers the one buildLoginCard appends.
cards = append(cards, randomComment(cs.env.Rand))
}
return &xfer.Message{Cards: cards}, nil
}
// buildLoginCard appends a random comment to cards and computes the login card
// over the result. It returns the login card and the card list it signed, which
// the caller must send verbatim behind that login card — the server recomputes
// the nonce from those bytes.
func (cs *cloneSession) buildLoginCard(cards []xfer.Card) (*xfer.LoginCard, []xfer.Card, error) {
signed := append(cards[:len(cards):len(cards)], randomComment(cs.env.Rand))
payload, err := encodeCards(signed)
if err != nil {
return nil, nil, err
}
login := computeLogin(cs.opts.User, cs.opts.Password, cs.projectCode, payload)
// BUGGIFY: 5% chance corrupt login nonce to test auth failure recovery.
if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.buildRequest.badLogin", 0.05) {
login.Nonce = "corrupted-nonce"
}
return login, signed, nil
}
// processResponse handles all cards in a server response for a clone round.
// cycle is that round's 0-based index; round 0 is the one sent before the
// project code — and so the login card — is known.
func (cs *cloneSession) processResponse(ctx context.Context, cycle int, msg *xfer.Message) error {
if msg == nil {
panic("sync.Clone.processResponse: msg must not be nil")
}
filesRecvd := 0
cs.roundPush = false
defer func() { cs.roundRecvd = filesRecvd }()
cardLoop:
for i, card := range msg.Cards {
if i%processResponseCancelCheckStride == 0 {
select {
case <-ctx.Done():
cs.result.BlobsRecvd += filesRecvd
return ctx.Err()
default:
}
}
switch c := card.(type) {
// §5.2: a reply card the decoder could not read ends the reply-card
// loop. On a clone it is a protocol defect, not a transport fault, so
// it aborts the clone without a retry (a retry would only hide it).
case *xfer.BadCommandCard:
return fmt.Errorf("sync.Clone: bad command in reply: %q", strings.TrimRight(c.Line, "\n"))
case *xfer.BadFileCard:
return fmt.Errorf("sync.Clone: %s", c.Message)
case *xfer.PushCard:
cs.roundPush = true
if err := cs.recordCodes(c); err != nil {
return err
}
case *xfer.FileCard:
if err := cs.receiveContent(ctx, c.UUID, c.DeltaSrc, c.Content, nil); err != nil {
return err
}
filesRecvd++
case *xfer.CFileCard:
stored, err := cs.receiveCFile(ctx, c)
if err != nil {
return err
}
if stored {
filesRecvd++
}
case *xfer.CloneSeqNoCard:
cs.recordSeqNo(c)
case *xfer.ErrorCard:
if cs.serverError(cycle, c) {
break cardLoop
}
case *xfer.CookieCard:
cs.cookie = c.Value
case *xfer.MessageCard:
cs.result.Messages = append(cs.result.Messages, c.Message)
}
}
cs.result.BlobsRecvd += filesRecvd
return nil
}
// receiveCFile stores a cfile and reports whether it did. fossil drops a
// shunned cfile even in a clone, though not a shunned file card (xfer.c:162,
// !cloneFlag), so a shunned one is skipped and not counted.
func (cs *cloneSession) receiveCFile(ctx context.Context, c *xfer.CFileCard) (bool, error) {
shunned, err := shun.IsShunned(cs.repo.DB(), c.UUID)
if err != nil || shunned {
return false, err
}
err = cs.receiveContent(ctx, c.UUID, c.DeltaSrc, c.Content, c.StoredBlob)
return err == nil, err
}
// recordSeqNo records the clone sequence a clone_seqno card carries. The
// server owns this cursor; the client only reads it. A non-decimal NEXT never
// reaches here — the decoder withholds it per §8.2 so the recorded sequence
// stays unchanged — and §8.2 bars the client from validating that a positive
// value advances.
func (cs *cloneSession) recordSeqNo(c *xfer.CloneSeqNoCard) {
// BUGGIFY: 5% chance ignore completion signal, forcing extra round.
if c.SeqNo == 0 && cs.opts.Buggify != nil &&
cs.opts.Buggify.Check("clone.processResponse.dropSeqNo", 0.05) {
return
}
cs.seqno = c.SeqNo
}
// serverError handles an error card and reports whether it ends the
// reply-card loop. A clone's first round carries no login card: the
// signature is salted with the project code, and the project code is exactly
// what this round exists to learn. A server that requires authentication
// therefore answers round 0 with its push card — which supplies that code —
// followed by "not authorized to clone". Treating that as fatal ends the
// clone before the credentials are ever offered, so round 0 records the
// message and carries on; round 1 repeats the request with a login card.
// Canonical does the same: `(syncFlags & SYNC_CLONE)==0 || nCycle>0` guards
// the abort in xfer.c client_sync.
//
// After round 0 an error ends the reply-card loop (§10.1): the cards after it
// are not read, and the session continues by §3.7 on what was stored before
// it. Clone still fails once the session ends, naming the error, so a clone
// cut short by the server is never reported complete.
func (cs *cloneSession) serverError(cycle int, c *xfer.ErrorCard) bool {
if c == nil {
panic("cloneSession.serverError: nil card")
}
cs.result.Messages = append(cs.result.Messages, c.Message)
if cycle == 0 {
return false
}
if cs.serverErr == "" {
cs.serverErr = c.Message
}
return true
}
// recordCodes stores the project and server codes a push card carries, the
// first time each arrives.
func (cs *cloneSession) recordCodes(c *xfer.PushCard) error {
if c == nil {
panic("cloneSession.recordCodes: nil card")
}
if c.ProjectCode != "" && cs.projectCode == "" {
cs.projectCode = c.ProjectCode
if _, err := cs.repo.DB().Exec(
"REPLACE INTO config(name, value) VALUES('project-code', ?)",
c.ProjectCode,
); err != nil {
return fmt.Errorf("sync.Clone: store project-code: %w", err)
}
}
if c.ServerCode != "" && cs.serverCode == "" {
cs.serverCode = c.ServerCode
if _, err := cs.repo.DB().Exec(
"REPLACE INTO config(name, value) VALUES('server-code', ?)",
c.ServerCode,
); err != nil {
return fmt.Errorf("sync.Clone: store server-code: %w", err)
}
}
return nil
}
// receiveContent stores the content a file or cfile card delivered.
// storedBlob is the cfile's payload in fossil's on-disk format, or nil.
func (cs *cloneSession) receiveContent(
ctx context.Context, uuid, deltaSrc string, content, storedBlob []byte,
) error {
// BUGGIFY: 2% chance corrupt file content to test hash verification.
// Relies on blob.Store verify-before-commit to catch corruption.
// storedBlob must be dropped alongside content: it decodes to the
// *original* bytes, so keeping it would let verbatim storage quietly mask
// the corruption this path exists to exercise, instead of the corrupted
// content being what gets hash-checked and stored.
if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.processResponse.corruptHash", 0.02) {
corrupted := make([]byte, len(content))
copy(corrupted, content)
if len(corrupted) > 0 {
corrupted[0] ^= 0xff
}
content = corrupted
storedBlob = nil
}
// BUGGIFY: 5% chance skip storing a received file, creating a phantom.
if cs.opts.Buggify != nil && cs.opts.Buggify.Check("clone.processResponse.dropFile", 0.05) {
return nil
}
return cs.handleFile(ctx, uuid, deltaSrc, content, storedBlob)
}
// handleFile stores a received file. storeReceivedFile now persists a
// delta whose base hasn't arrived yet rather than discarding it (see
// storeDeltaAgainstPhantomBase), so uuid itself is never re-requested here
// once stored — only a still-missing base might need another round.
//
// storedBlob, when non-nil, is payload already re-expressed in Fossil's
// on-disk blob format (see xfer.CFileCard.StoredBlob) and is persisted
// verbatim rather than recompressed — this is what makes a clone
// byte-identical to its source's blob table, not merely content-identical.
func (cs *cloneSession) handleFile(ctx context.Context, uuid, deltaSrc string, payload []byte, storedBlob []byte) error {
if err := storeReceivedFile(ctx, cs.repo, uuid, deltaSrc, payload, storedBlob, visibilityPublic, cs.linker); err != nil {
return fmt.Errorf("sync.Clone: handleFile %s: %w", uuid, err)
}
delete(cs.phantoms, uuid)
if deltaSrc != "" {
if _, cached := cs.availableDeltaSources[deltaSrc]; !cached {
// Availability, not existence: a phantom row for deltaSrc still
// needs requesting, and so does a delta whose own base is a
// phantom. Only a confirmed available source is memoized: an
// unavailable source can arrive later in this same clone.
if _, available := content.AvailableByUUID(cs.repo.DB(), deltaSrc); !available {
// request_phantoms never asks for a shunned name.
shunned, err := shun.IsShunned(cs.repo.DB(), deltaSrc)
if err != nil {
return fmt.Errorf("sync.Clone: basis of %s: %w", uuid, err)
}
if !shunned {
cs.phantoms[deltaSrc] = true
}
} else if len(cs.availableDeltaSources) < cloneAvailableDeltaSourceLimit {
cs.availableDeltaSources[deltaSrc] = struct{}{}
}
}
}
return nil
}