Move probe and fetch shaping into session helpers · Entire
Move probe and fetch shaping into session helpers
aca59be→main·
Soph·3mo ago·2 files·+90 added/-67 removed
Sessions
Changes
2
docs
Mrewrite-issue-list.md+1
internal/syncer
Msyncer.go+89/-67
300 unmodified lines
301
302
303
304
305
306
307
300 unmodified lines
- Syncer bootstrap execution now takes a shared session object instead of threading raw source/target transports, stats, logger, measurement, and target advertisement through a wide helper signature.
- Syncer normal execution now routes both incremental relay and materialized fallback through session-owned helpers instead of rebuilding transport, capability, and policy inputs inline inside `Run`.
- Syncer target-side state is now grouped into a dedicated target session object so advertised refs, derived features, relay policy, and the push executor move together instead of being carried as parallel top-level fields.
- Syncer source-side probe/fetch result assembly and have-ref normalization now also live behind session-owned helpers instead of rebuilding those structures ad hoc in the public entrypoints.
- Some helpers still carry broad parameter structs, so this remains partial rather than fully complete.
## Performance And Scalability
Mdocs/rewrite-issue-list.md+1
544 unmodified lines
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
548
549
550
551
11 unmodified lines
563
564
565
590
591
592
566
567
568
569
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
570
571
572
573
574
575
3 unmodified lines
579
580
581
620
621
622
623
624
625
582
583
584
585
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
586
587
588
589
52 unmodified lines
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
544 unmodified lines
if err != nil {
return ProbeResult{}, err
}
refInfos := make([]RefInfo, 0, len(s.sourceRefMap))
for name, hash := range s.sourceRefMap {
refInfos = append(refInfos, RefInfo{Name: name.String(), Hash: hash})
}
sort.Slice(refInfos, func(i, j int) bool { return refInfos[i].Name < refInfos[j].Name })
result := ProbeResult{
SourceURL: cfg.Source.URL,
RequestedMode: cfg.ProtocolMode,
Protocol: s.sourceService.Protocol,
RefPrefixes: planner.RefPrefixes(cfg.Mappings, cfg.IncludeTags),
Capabilities: s.sourceService.Capabilities(),
Refs: refInfos,
Stats: s.stats.snapshot(),
Measurement: s.measurementDone(),
}
if cfg.Target.URL != "" {
result.TargetURL = cfg.Target.URL
result.TargetCaps = gitproto.AdvRefsCaps(s.target.adv)
result.Stats = s.stats.snapshot()
result.Measurement = s.measurementDone()
}
return result, nil
return s.newProbeResult(), nil
}
// Fetch exercises source-side fetch negotiation.
11 unmodified lines
if err != nil {
return FetchResult{}, fmt.Errorf("init in-memory repository: %w", err)
}
sourceRefMap := s.sourceRefMap
desiredRefs, _, err := planner.BuildDesiredRefs(sourceRefMap, planConfig(cfg))
desiredRefs, err := s.buildDesiredRefs()
if err != nil {
return FetchResult{}, err
}
if len(desiredRefs) == 0 {
return FetchResult{}, fmt.Errorf("no source refs matched")
}
targetRefMap := make(map[plumbing.ReferenceName]plumbing.Hash)
for _, raw := range haveRefs {
name := validation.ParseHaveRef(raw)
hash, ok := sourceRefMap[name]
if !ok {
return FetchResult{}, fmt.Errorf("have-ref %q not found on source", raw)
}
targetRefMap[name] = hash
}
for idx, hash := range haveHashes {
targetRefMap[plumbing.ReferenceName(fmt.Sprintf("refs/haves/%%d", idx))] = hash
targetRefMap, err := s.buildHaveRefMap(haveRefs, haveHashes)
if err != nil {
return FetchResult{}, err
}
gpDesired := convert.DesiredRefs(desiredRefs)
3 unmodified lines
}
}
wants := make([]RefInfo, 0, len(desiredRefs))
for _, ref := range desiredRefs {
wants = append(wants, RefInfo{Name: ref.SourceRef.String(), Hash: ref.SourceHash})
}
sort.Slice(wants, func(i, j int) bool { return wants[i].Name < wants[j].Name })
objectCount, err := countObjects(repo.Storer)
if err != nil {
return FetchResult{}, fmt.Errorf("count fetched objects: %w", err)
}
haveValues := make([]plumbing.Hash, 0, len(targetRefMap))
for _, h := range targetRefMap {
if !h.IsZero() {
haveValues = append(haveValues, h)
}
}
return FetchResult{
SourceURL: cfg.Source.URL,
RequestedMode: cfg.ProtocolMode,
Protocol: s.sourceService.Protocol,
Wants: wants,
Haves: gitproto.SortedUniqueHashes(haveValues),
FetchedObjects: objectCount,
Stats: s.stats.snapshot(),
Measurement: s.measurementDone(),
}, nil
return s.newFetchResult(objectCount, desiredRefs, targetRefMap), nil
}
// --- Bootstrap implementation ---
52 unmodified lines
})
}
func (s *syncSession) buildDesiredRefs() (map[plumbing.ReferenceName]planner.DesiredRef, error) {
desiredRefs, _, err := planner.BuildDesiredRefs(s.sourceRefMap, planConfig(s.cfg))
if err != nil {
return nil, err
}
if len(desiredRefs) == 0 {
return nil, fmt.Errorf("no source refs matched")
}
return desiredRefs, nil
}
func (s *syncSession) buildHaveRefMap(haveRefs []string, haveHashes []plumbing.Hash) (map[plumbing.ReferenceName]plumbing.Hash, error) {
targetRefMap := make(map[plumbing.ReferenceName]plumbing.Hash)
for _, raw := range haveRefs {
name := validation.ParseHaveRef(raw)
hash, ok := s.sourceRefMap[name]
if !ok {
return nil, fmt.Errorf("have-ref %q not found on source", raw)
}
targetRefMap[name] = hash
}
for idx, hash := range haveHashes {
targetRefMap[plumbing.ReferenceName(fmt.Sprintf("refs/haves/%%d", idx))] = hash
}
return targetRefMap, nil
}
func (s *syncSession) newProbeResult() ProbeResult {
refInfos := make([]RefInfo, 0, len(s.sourceRefMap))
for name, hash := range s.sourceRefMap {
refInfos = append(refInfos, RefInfo{Name: name.String(), Hash: hash})
}
sort.Slice(refInfos, func(i, j int) bool { return refInfos[i].Name < refInfos[j].Name })
result := ProbeResult{
SourceURL: s.cfg.Source.URL,
RequestedMode: s.cfg.ProtocolMode,
Protocol: s.sourceService.Protocol,
RefPrefixes: planner.RefPrefixes(s.cfg.Mappings, s.cfg.IncludeTags),
Capabilities: s.sourceService.Capabilities(),
Refs: refInfos,
Stats: s.stats.snapshot(),
Measurement: s.measurementDone(),
}
if s.target != nil {
result.TargetURL = s.cfg.Target.URL
result.TargetCaps = gitproto.AdvRefsCaps(s.target.adv)
result.Stats = s.stats.snapshot()
result.Measurement = s.measurementDone()
}
return result
}
func (s *syncSession) newFetchResult(
objectCount int,
desiredRefs map[plumbing.ReferenceName]planner.DesiredRef,
targetRefMap map[plumbing.ReferenceName]plumbing.Hash,
) FetchResult {
wants := make([]RefInfo, 0, len(desiredRefs))
for _, ref := range desiredRefs {
wants = append(wants, RefInfo{Name: ref.SourceRef.String(), Hash: ref.SourceHash})
}
sort.Slice(wants, func(i, j int) bool { return wants[i].Name < wants[j].Name })
haveValues := make([]plumbing.Hash, 0, len(targetRefMap))
for _, h := range targetRefMap {
if !h.IsZero() {
haveValues = append(haveValues, h)
}
}
return FetchResult{
SourceURL: s.cfg.Source.URL,
RequestedMode: s.cfg.ProtocolMode,
Protocol: s.sourceService.Protocol,
Wants: wants,
Haves: gitproto.SortedUniqueHashes(haveValues),
FetchedObjects: objectCount,
Stats: s.stats.snapshot(),
Measurement: s.measurementDone(),
}
}
func countObjects(store storer.EncodedObjectStorer) (int, error) {
iter, err := store.IterEncodedObjects(plumbing.AnyObject)
if err != nil {