Artifact [24245c243d]

Artifact 24245c243d38908b665b1ef7bc9eb2cf2ab12b7a6b5603c883fe5cdc79c0c415:

• File conformance/sparsesync_helpers_test.go — part of check-in [352a0d9e0b] at 2026-10-22 12:02:50 on branch feat/horizon-cones — feat(sync): serve the content horizon and send its pragma

  A horizon store asked every peer for the whole history and threw away what fell below C on arrival, so a horizon clone cost a dense clone's bandwidth. draft-fossil-horizon-00 §6 lets the request carry the horizon and the origin restrict what it sends.

  The client sends `pragma horizon HASH RULE` after its path-cone pragmas on every clone and pull, never horizon-structural, and counts the replies that say `pragma horizon-served content` (HorizonServed on the clone and sync results, internal and public). It still classifies every reply locally, so a horizon-ignorant peer stays correct (§6.3).

  The handler reads the pragma into the request's cone and serves it through the existing keep machinery rather than a parallel path: the pull inventory's keep set drops the blobs below the horizon (via ConeReferences.HorizonBelow) and every cluster; the clone walk walks the tree of a held-set check-in only, counts a parent as sent only inside the held set, and sends every sub-manifest node at its row while sending no cluster. A crossing delta is inflated through the same sourceKept test the path cone uses, charged to the round's batch. req-clusters under a horizon answers nothing, and every reply says horizon-served content.

  A horizon store refuses, through refuseClone, a dense clone and a horizon whose held set is not inside its own, both computed over the store with horizon.Place. That is the spec's subset test; zit refuses any other C or rule. A path-only store now also refuses a horizon-only request, which keeps every path.

  Interop with zit at horizon-rfc bbd1435674 holds both ways: zit clones and pulls at a horizon from the handler, the client from `zit serve`, and each side's horizon store refuses the other's dense clone. (user: [email protected], size: 3985) [annotate] [blame] [check-ins using] [more...]

package conformance

import (
	"context"
	"fmt"
	"path/filepath"
	"testing"
	"time"

	"github.com/danmestas/go-libfossil/internal/content"
	"github.com/danmestas/go-libfossil/internal/hash"
	"github.com/danmestas/go-libfossil/internal/repo"
	"github.com/danmestas/go-libfossil/internal/sync"
	"github.com/danmestas/go-libfossil/internal/xfer"
	"github.com/danmestas/go-libfossil/simio"
)

// scSyncPeer records decoded wire transcripts, including requests an old peer
// ignores. Both directions cross the actual xfer encoder and decoder.
type scSyncPeer struct {
	r        *repo.Repo
	ignore   bool
	opts     sync.HandleOpts
	requests []*xfer.Message
	replies  []*xfer.Message
}

func scSyncWire(m *xfer.Message) (*xfer.Message, error) {
	data, err := m.Encode()
	if err != nil {
		return nil, err
	}
	return xfer.Decode(data, xfer.ContentTypeCompressed)
}

func (p *scSyncPeer) Exchange(ctx context.Context, req *xfer.Message) (*xfer.Message, error) {
	req, err := scSyncWire(req)
	if err != nil {
		return nil, err
	}
	p.requests = append(p.requests, req)
	if p.ignore {
		cards := make([]xfer.Card, 0, len(req.Cards))
		for _, card := range req.Cards {
			if c, ok := card.(*xfer.PragmaCard); ok && (c.Name == "cone" || c.Name == "cone-not" || c.Name == "horizon") {
				continue
			}
			cards = append(cards, card)
		}
		req = &xfer.Message{Cards: cards}
	}
	resp, err := sync.HandleSyncWithOpts(ctx, p.r, req, p.opts)
	if err != nil {
		return nil, err
	}
	resp, err = scSyncWire(resp)
	if err != nil {
		return nil, err
	}
	p.replies = append(p.replies, resp)
	return resp, nil
}

func newSCSyncRepo(t *testing.T, files map[string]string) *scRepo {
	t.Helper()
	r, err := repo.Create(filepath.Join(t.TempDir(), "origin.fossil"), "u", simio.CryptoRand{}, "")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { r.Close() })
	if err := r.SetHashPolicy("sha1"); err != nil {
		t.Fatal(err)
	}
	if _, err := r.DB().Exec("UPDATE user SET cap='gio' WHERE login='nobody'"); err != nil {
		t.Fatal(err)
	}
	f := &scRepo{r: r, when: time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)}
	f.c1 = f.commit(t, files, 0)
	return f
}

func scSyncFiles() map[string]string {
	return map[string]string{
		"src/a": "inside\n", "src/shared": "shared\n",
		"zzz/shared": "shared\n", "zzz/b": "outside\n",
	}
}

func scSyncMember(value string) []xfer.Card {
	return []xfer.Card{&xfer.PragmaCard{Name: "cone", Values: []string{value}}}
}

func scSyncClone(
	t *testing.T, peer *scSyncPeer, opts sync.CloneOpts,
) (*repo.Repo, *sync.CloneResult) {
	t.Helper()
	r, result, err := sync.Clone(context.Background(),
		filepath.Join(t.TempDir(), "clone.fossil"), peer, opts)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { r.Close() })
	return r, result
}

func scSyncHolding(r *repo.Repo, text string, want bool) error {
	uuid := hash.SHA1([]byte(text))
	_, got, err := content.CheckAvailableByUUID(r.DB(), uuid)
	if err != nil {
		return err
	}
	if got != want {
		return fmt.Errorf("holding %s = %v, want %v", uuid, got, want)
	}
	return nil
}

func scSyncNoGimme(messages []*xfer.Message, uuid string) error {
	for _, m := range messages {
		for _, card := range m.Cards {
			if c, ok := card.(*xfer.GimmeCard); ok && c.UUID == uuid {
				return fmt.Errorf("requested elided artifact %s", uuid)
			}
		}
	}
	return nil
}

func scSyncPayloads(m *xfer.Message) map[string]string {
	out := map[string]string{}
	for _, card := range m.Cards {
		switch c := card.(type) {
		case *xfer.FileCard:
			out[c.UUID] = c.DeltaSrc
		case *xfer.CFileCard:
			out[c.UUID] = c.DeltaSrc
		}
	}
	return out
}

func scSyncRefused(m *xfer.Message, want string) error {
	var errors []string
	for _, card := range m.Cards {
		if c, ok := card.(*xfer.ErrorCard); ok {
			errors = append(errors, c.Message)
		}
	}
	if len(errors) != 1 || errors[0] != want || len(scSyncPayloads(m)) != 0 {
		return fmt.Errorf("refusal errors=%q payloads=%v; want one %q and no content",
			errors, scSyncPayloads(m), want)
	}
	return nil
}