Artifact [f836aac8a6]

Artifact f836aac8a66eb7df4e7c4fa1c68c15f51f205a33ce3ae10122a33dda6a3ef68e:

• File internal/sync/session.go — part of check-in [a565d1c467] at 2026-10-13 14:49:38 on branch feat/425-share-links — feat(sync): add share-links mode for pragma req-links and link

  sync-protocol-01 Section 9.5 makes share-links mode OPTIONAL. With SyncOpts.ShareLinks a pulling session sends pragma req-links in its first request, as fossil's --share-links does, unless SyncOpts.ProjectCode names another project. It keeps each pragma link URL ARG MTIME as config link:URL = ARG, defossilized and otherwise verbatim, replacing an existing link only with a strictly newer MTIME. A wrong token count, a URL that is not HTTP(S) or an MTIME that is not a positive decimal drops the link.

  Where fossil 2.28 differs the draft is followed: fossil fails fatally on a link URL its parser rejects and reads MTIME with strtoll base 0.

  New sync.9.5 client cases run against fossil sync --share-links as well. SP-9.5-4 loses its scope note and is covered; SP-9.5-5 and SP-9.5-6 are new rows for which links a client keeps and when it sends req-links.

  Tracked in #425. (user: [email protected], size: 15570) [annotate] [blame] [check-ins using] [more...]

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
	ShareLinks   bool           // share-links mode: request and keep links (§9.5); pull only
	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; rid-keyed, so not across a rollback
}

// 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
	// pass is the current reply's verification pass (§7.7).
	pass             *receivePass
	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.
	// fossil runs manifest_crosslink_end inside each round's transaction
	// (xfer.c); here the repair runs once, after the rounds, in its own
	// transaction, so that a cancelled sweep cannot roll back a committed
	// reply. A rolled-back reply restores s.cascaded, so none is owed for it.
	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
}