Artifact [e851fc7c9e]

Artifact e851fc7c9e4fe8b49c426efa01d47e1c5f0b513c936219ca62b1d510cf8a3737:

• File internal/sync/session.go — part of check-in [0035af8ac5] at 2026-10-13 12:27:17 on branch main — fix(sync): repair leaves and tags for check-ins the cascade links (#414) (#416)

  * fix(sync): repair leaves and tags for check-ins the cascade links (#414)

  A pulled check-in arrives as an announced phantom filled the next round, so the phantom-fill cascade links it. The session's closing sweep then found nothing left to link and skipped the leaf and tag-propagation repair, which gated on the sweep's own count: the check-in had an event row and no leaf row, and its parent stayed a leaf. The server handler had the same gap for a pushed artifact that fills a phantom.

  AfterDephantomize and AfterDephantomizeBatch now return how many artifacts they linked. The sync client and the handler add that up over the session and close it with the new CrosslinkAfterCascades, which runs the repair once when the sweep or the cascades linked anything (repo-state-01 Section 7.3, RS-7.3-4), not once per file. The receive linker's cancellation paths gate on the same flag.

  The repo-state harness gains a pull load: the first artifacts of a script are held and crosslinked locally, the rest pulled in process from a go-libfossil server holding them all. New cases repostate.7.3.pulled-child and repostate.7.6.pulled-retag, checked against fossil per artifact, each run as a .pull case too.

  Tracked in #414.

  * fix(sync): repair leaves of cascade links when a session fails (#414) (user: [email protected], size: 15112) [annotate] [blame] [check-ins using]

package sync

import (
	"context"
	"errors"
	"fmt"

	"github.com/danmestas/go-libfossil/internal/content"
	libfossil "github.com/danmestas/go-libfossil/internal/fsltype"
	"github.com/danmestas/go-libfossil/internal/manifest"
	"github.com/danmestas/go-libfossil/internal/repo"
	"github.com/danmestas/go-libfossil/internal/uv"
	"github.com/danmestas/go-libfossil/internal/xfer"
	"github.com/danmestas/go-libfossil/simio"
)

const (
	// DefaultMaxSend is the default byte budget per round for file payloads.
	DefaultMaxSend = 250000
	// MaxRounds caps the number of sync rounds before giving up.
	MaxRounds = 100
	// MaxGimmeBase is the minimum gimme cap per round.
	MaxGimmeBase = 200
)

// BuggifyChecker is an optional fault injection interface.
// Pass nil in production — implementations should be nil-safe.
type BuggifyChecker interface {
	Check(site string, probability float64) bool
}

// SyncOpts configures a sync session.
type SyncOpts struct {
	Push, Pull   bool           // enable push/pull directions (at least one must be true)
	ProjectCode  string         // repo's project-code — sent to identify the repository
	ServerCode   string         // server-code from a previous session (cookie-like, speeds up sync)
	User         string         // login user — empty means unauthenticated "nobody" sync
	Password     string         // login password
	PeerID       string         // identifies this leaf agent instance (for observability)
	MaxSend      int            // byte budget per round for file payloads (0 defaults to DefaultMaxSend)
	UV           bool           // enable unversioned file sync (wiki, forum, attachments)
	XTableSync   bool           // enable extension table sync (peer_registry, etc.) — only between EdgeSync peers
	Private      bool           // enable private artifact sync
	Env          *simio.Env     // nil defaults to RealEnv
	Buggify      BuggifyChecker // nil in production — used by DST for fault injection
	Observer     Observer       // nil defaults to no-op
	CkinLock     *CkinLockReq   // nil = no lock requested
	ContentCache *content.Cache // nil = no caching (every Expand walks the full delta chain)
}

// CkinLockReq requests a server-side check-in lock.
type CkinLockReq struct {
	ParentUUID string
	ClientID   string
}

// SyncResult reports what happened during a sync.
type SyncResult struct {
	Rounds, FilesSent, FilesRecvd int
	UVFilesSent, UVFilesRecvd     int
	UVGimmesSent                  int
	BytesSent, BytesRecvd         int64
	ArtifactsLinked               int
	Errors                        []string
	CkinLockFail                  *CkinLockFail // nil = no conflict
}

// session holds the mutable state of a running sync.
type session struct {
	repo              *repo.Repo
	env               *simio.Env
	opts              SyncOpts
	projectCode       string // resolved once: SyncOpts.ProjectCode, else the repo's own
	result            SyncResult
	cookie            string
	remoteHas         map[string]bool
	phantoms          map[string]bool
	pendingSend       map[string]bool
	mxPhantomReq      int // gimme cards allowed per request; fossil's mxPhantomReq
	igotSentThisRound int
	maxSend           int
	phantomAge        map[string]int // UUID -> consecutive rounds gimme'd without delivery
	uvHashSent        bool
	uvPushOK          bool
	uvPullOnly        bool
	uvToSend          map[string]bool // name -> true=full content, false=mtime-only
	uvGimmes          map[string]bool
	nUvGimmeSent      int
	nUvFileRcvd       int
	nGimmeRcvd        int  // cumulative gimmes received across all rounds
	nextIsPrivate     bool // true when a PrivateCard precedes the next file/cfile
	// pushStopped is set the moment the server's "message pull only ..." is
	// read (§10.2): the client keeps pulling but answers no further gimme, as
	// fossil's client_sync clears SYNC_PUSH mid-reply, so a gimme after the
	// message in the same reply goes unanswered.
	pushStopped bool
	// pushCardDropped lags pushStopped by one request. fossil emits the next
	// request's push card before it reads the reply that stops the push, so
	// the request built right after the message still carries its push card;
	// only the one after that drops it. buildRequest promotes pushStopped into
	// this after emitting each request's push card.
	pushCardDropped bool
	// visibilityUpdates buffers this round's igot-announced status changes so
	// they are written in one transaction rather than one per card.
	visibilityUpdates []visibilityUpdate
	// newPhantoms buffers this round's igot phantoms for the same reason.
	newPhantoms []phantomUpdate
	// peerVersion is the server's release from server-version; before one
	// arrives it is this library's own, as in fossil's client_sync (§7.6).
	peerVersion int
	// loginCookieOK is set once a server-version passes the §6.5 gate:
	// later requests may then log in through the x-f-l-c cookie.
	loginCookieOK bool
	// sha3Refusals counts gimmes for SHA3 content withheld from an older
	// server; the next request carries one refusal error card for each.
	sha3Refusals int
	// peerDate is the server's check-in date from server-version: 0 before
	// one arrives, or when it carries no DATE and TIME, which no date gate
	// passes (§8.6).
	peerDate int
	// privateIGots are the private artifacts a gimme asked for outside a
	// private sync; the next request answers each with igot HASH 1 (§8.6).
	privateIGots []string
	// dephantomized buffers the phantoms this reply filled, whose
	// after-dephantomize cascade runs once at the end of the reply.
	dephantomized []libfossil.FslID
	// cascaded counts what those cascades linked, across the session: the
	// closing crosslink owes it the leaf and tag repair (issue #414).
	cascaded         int
	roundStats       RoundStats
	xTableHashSent   map[string]bool            // table -> true if xtable-hash pragma sent
	xTableGimmes     map[string]map[string]bool // table -> pkHash -> true
	xTableToSend     map[string]map[string]bool // table -> pkHash -> true
	xTableCache      map[string]*repo.TableDef  // cached table defs, lazily loaded
	dephantomizeHook func(libfossil.FslID)      // called after phantom→real transition
	cache            *content.Cache             // nil = passthrough to content.Expand
}

func newSession(r *repo.Repo, opts SyncOpts) *session {
	if r == nil {
		panic("sync.newSession: r must not be nil")
	}
	ms := opts.MaxSend
	if ms <= 0 {
		ms = DefaultMaxSend
	}
	env := opts.Env
	if env == nil {
		env = simio.RealEnv()
	}
	// A repo knows its own project code, so a caller holding one never has to
	// supply it. An explicit SyncOpts.ProjectCode still wins; a repo with no
	// stored code leaves this empty, and the sites that need one say so.
	projectCode := opts.ProjectCode
	if projectCode == "" {
		projectCode, _ = r.Config("project-code")
	}

	s := &session{
		repo:           r,
		env:            env,
		opts:           opts,
		projectCode:    projectCode,
		maxSend:        ms,
		remoteHas:      make(map[string]bool),
		phantoms:       make(map[string]bool),
		mxPhantomReq:   MaxGimmeBase,
		pendingSend:    make(map[string]bool),
		phantomAge:     make(map[string]int),
		peerVersion:    releaseVersion,
		xTableHashSent: make(map[string]bool),
		xTableGimmes:   make(map[string]map[string]bool),
		xTableToSend:   make(map[string]map[string]bool),
		cache:          opts.ContentCache,
	}

	// Pre-populate uvToSend with the local files, tombstones left out, as
	// fossil seeds uv_tosend from rows whose hash is not NULL (§8.5.1). A
	// local deletion newer than the server's copy is queued when its
	// uvigot shows that.
	if opts.UV {
		if err := uv.EnsureSchema(r.DB()); err != nil {
			panic(fmt.Sprintf("sync.newSession: uv.EnsureSchema: %v", err))
		}
		entries, err := uv.List(r.DB())
		if err != nil {
			panic(fmt.Sprintf("sync.newSession: uv.List: %v", err))
		}
		s.uvToSend = make(map[string]bool)
		s.uvGimmes = make(map[string]bool)
		for _, e := range entries {
			if e.Hash != "" {
				s.uvToSend[e.Name] = true
			}
		}
	}

	return s
}

// errNoProjectCode reports that neither the caller nor the repo could supply
// the project code the named wire element needs.
func errNoProjectCode(what string) error {
	return fmt.Errorf("sync: %s requires a project-code: none in SyncOpts.ProjectCode and none stored in the repo's config", what)
}

// getXTableDef returns a cached table definition, loading all tables on first miss.
func (s *session) getXTableDef(tableName string) (*repo.TableDef, error) {
	if s.xTableCache != nil {
		if def, ok := s.xTableCache[tableName]; ok {
			return def, nil
		}
	}
	// Cache miss — load all tables.
	tables, err := repo.ListSyncedTables(s.repo.DB())
	if err != nil {
		return nil, err
	}
	s.xTableCache = make(map[string]*repo.TableDef)
	for _, info := range tables {
		d := info.Def
		s.xTableCache[info.Name] = &d
	}
	if def, ok := s.xTableCache[tableName]; ok {
		return def, nil
	}
	return nil, nil
}

// Sync runs the client sync loop against the given transport.
// It returns once the protocol has converged or a fatal error occurs.
func Sync(ctx context.Context, r *repo.Repo, t Transport, opts SyncOpts) (result *SyncResult, err error) {
	if r == nil {
		panic("sync.Sync: r must not be nil")
	}
	if t == nil {
		panic("sync.Sync: t must not be nil")
	}
	defer func() {
		// The invariant below is only meaningful on a normal return: while a
		// panic unwinds, err and result are both nil and asserting here would
		// replace the real cause with this one.
		if p := recover(); p != nil {
			panic(p)
		}
		if err == nil && result == nil {
			panic("sync.Sync: result must not be nil on success")
		}
	}()

	obs := resolveObserver(opts.Observer)
	s := newSession(r, opts)
	closed := false
	defer func() {
		// A round that fails after earlier rounds' cascades committed their
		// links must still repair leaves and propagated tags, as fossil's
		// per-round manifest_crosslink_end does (src/xfer.c): otherwise a
		// failed pull leaves check-ins with event rows and no leaf rows
		// (repo-state-01 Section 7.3, RS-7.3-4). The context is detached so a
		// cancelled pull is repaired too.
		if closed || err == nil || s.cascaded == 0 {
			return
		}
		_, rerr := manifest.CrosslinkAfterCascades(context.WithoutCancel(ctx), s.repo, s.cascaded)
		if rerr != nil {
			err = errors.Join(err, fmt.Errorf("sync: repair after failed session: %w", rerr))
		}
	}()
	if err := s.seedPhantoms(); err != nil {
		return &s.result, fmt.Errorf("sync: load phantoms: %w", err)
	}

	ctx = obs.Started(ctx, SessionStart{
		Operation:   "sync",
		Push:        opts.Push,
		Pull:        opts.Pull,
		UV:          opts.UV,
		ProjectCode: s.projectCode,
		PeerID:      opts.PeerID,
	})

	for cycle := 0; ; cycle++ {
		select {
		case <-ctx.Done():
			obs.Completed(ctx, sessionEndFromSync(&s.result, s.projectCode), ctx.Err())
			return &s.result, ctx.Err()
		default:
		}
		if cycle >= MaxRounds {
			err := fmt.Errorf("sync: exceeded %d rounds", MaxRounds)
			obs.Completed(ctx, sessionEndFromSync(&s.result, s.projectCode), err)
			return &s.result, err
		}

		roundCtx := obs.RoundStarted(ctx, cycle)
		s.roundStats = RoundStats{}

		req, err := s.buildRequest(cycle)
		if err != nil {
			obs.Error(roundCtx, err)
			obs.RoundCompleted(roundCtx, cycle, s.roundStats)
			obs.Completed(ctx, sessionEndFromSync(&s.result, s.projectCode), err)
			return &s.result, fmt.Errorf("sync: buildRequest round %d: %w", cycle, err)
		}
		resp, err := exchangeWithRetry(ctx, t, req, s.env.Clock, obs)
		if err != nil {
			obs.Error(roundCtx, err)
			obs.RoundCompleted(roundCtx, cycle, s.roundStats)
			obs.Completed(ctx, sessionEndFromSync(&s.result, s.projectCode), err)
			return &s.result, fmt.Errorf("sync: exchange round %d: %w", cycle, err)
		}

		sentBefore := s.result.FilesSent
		recvdBefore := s.result.FilesRecvd

		done, err := s.processResponse(ctx, resp)

		s.roundStats.FilesSent = s.result.FilesSent - sentBefore
		s.roundStats.FilesReceived = s.result.FilesRecvd - recvdBefore
		s.result.BytesSent += s.roundStats.BytesSent
		s.result.BytesRecvd += s.roundStats.BytesReceived

		if err != nil {
			obs.Error(roundCtx, err)
			obs.RoundCompleted(roundCtx, cycle, s.roundStats)
			obs.Completed(ctx, sessionEndFromSync(&s.result, s.projectCode), err)
			return &s.result, fmt.Errorf("sync: processResponse round %d: %w", cycle, err)
		}
		s.result.Rounds = cycle + 1

		obs.RoundCompleted(roundCtx, cycle, s.roundStats)

		if done {
			break
		}
	}

	// Auto-crosslink after convergence. Context-aware for the same reason the
	// clone's sweep is (#120): this runs after the round loop, so it has no
	// round boundary of its own at which a cancelled context could be noticed.
	closed = true
	linked, xlinkErr := manifest.CrosslinkAfterCascades(ctx, s.repo, s.cascaded)
	if xlinkErr != nil {
		obs.Completed(ctx, sessionEndFromSync(&s.result, s.projectCode), xlinkErr)
		return &s.result, fmt.Errorf("sync: crosslink: %w", xlinkErr)
	}
	s.result.ArtifactsLinked = linked

	obs.Completed(ctx, sessionEndFromSync(&s.result, s.projectCode), nil)
	return &s.result, nil
}

// sessionEndFromSync builds a SessionEnd from a SyncResult.
func sessionEndFromSync(r *SyncResult, projectCode string) SessionEnd {
	return SessionEnd{
		Operation:    "sync",
		Rounds:       r.Rounds,
		FilesSent:    r.FilesSent,
		FilesRecvd:   r.FilesRecvd,
		UVFilesSent:  r.UVFilesSent,
		UVFilesRecvd: r.UVFilesRecvd,
		UVGimmesSent: r.UVGimmesSent,
		BytesSent:    r.BytesSent,
		BytesRecvd:   r.BytesRecvd,
		ProjectCode:  projectCode,
		Errors:       r.Errors,
	}
}

// cardSummary returns a short string describing a card (for trace logging).
// codePrefix returns the first 8 characters of a code for trace logging,
// or the full string when shorter. Avoids slice-out-of-range panics when
// ServerCode/ProjectCode are empty (the no-codes-negotiated form).
func codePrefix(s string) string {
	if len(s) >= 8 {
		return s[:8]
	}
	return s
}

func cardSummary(c xfer.Card) string {
	switch v := c.(type) {
	case *xfer.PullCard:
		return fmt.Sprintf("srv=%s proj=%s", codePrefix(v.ServerCode), codePrefix(v.ProjectCode))
	case *xfer.PushCard:
		return fmt.Sprintf("srv=%s proj=%s", codePrefix(v.ServerCode), codePrefix(v.ProjectCode))
	case *xfer.IGotCard:
		return v.UUID[:16] + "..."
	case *xfer.GimmeCard:
		return v.UUID[:16] + "..."
	case *xfer.FileCard:
		return fmt.Sprintf("uuid=%s len=%d delta=%s", v.UUID[:16], len(v.Content), v.DeltaSrc)
	case *xfer.CFileCard:
		return fmt.Sprintf("uuid=%s len=%d delta=%s", v.UUID[:16], len(v.Content), v.DeltaSrc)
	case *xfer.LoginCard:
		return fmt.Sprintf("user=%s", v.User)
	case *xfer.CookieCard:
		return v.Value
	case *xfer.ErrorCard:
		return v.Message
	case *xfer.MessageCard:
		return v.Message
	case *xfer.PragmaCard:
		return v.Name
	default:
		return ""
	}
}

// cardsByType is a helper that filters cards by type for testing/debugging.
func cardsByType(msg *xfer.Message, ct xfer.CardType) []xfer.Card {
	var out []xfer.Card
	for _, c := range msg.Cards {
		if c.Type() == ct {
			out = append(out, c)
		}
	}
	return out
}