Bundle target state in sync session

3d6606d→main · Soph · 3mo ago · 2 files · +38 added / -27 removed

Sessions

5683f5862e46 View transcript

Changes

2

• docs

• Mrewrite-issue-list.md +2 / -1

• internal/syncer

• Msyncer.go +36 / -26

299 unmodified lines
300
301
302
303
304
305
306
151 unmodified lines
458
459
460
460
461
462
463
464
299 unmodified lines
- The materialized fallback now runs through an explicit executor with separate stages for tag prefetch, object-closure collection, limit enforcement, and push execution.
- 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.
- Some helpers still carry broad parameter structs, so this remains partial rather than fully complete.

## Performance And Scalability
151 unmodified lines

- All mapping validation happens before network activity. Status: done
- Capability negotiation is centralized and enforced consistently. Status: partial
  Source-side fetch capability checks now live behind `gitproto.RefService` methods, and sync session setup now owns target advertisement, derived target features, relay policy, and the target pusher together instead of rebuilding those pieces ad hoc in `Run`. Some target-side relay decisions still rely on orchestration wiring rather than a fully unified capability model, so this remains partial.
  Source-side fetch capability checks now live behind `gitproto.RefService` methods, and sync session setup now owns target advertisement, derived target features, relay policy, and the target pusher together inside a dedicated target session instead of rebuilding those pieces ad hoc in `Run`. Some target-side relay decisions still rely on orchestration wiring rather than a fully unified capability model, so this remains partial.
- Relay strategies are separate packages with explicit inputs and outputs. Status: done
- Tag creation is correct whether or not a pack transfer is needed. Status: done
- Stats are concurrency-safe. Status: done

Mdocs/rewrite-issue-list.md +2 / -1

309 unmodified lines
310
311
312
313
313
315
316
317
318
314
320
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
31 unmodified lines
362
363
364
361
365
366
367
368
365
369
370
371
372
369
373
374
375
376
373
374
375
376
377
377
378
379
380
381
382
383
384
385
386
387
388
389
379
390
391
392
11 unmodified lines
404
405
406
397
407
408
409
410
60 unmodified lines
471
472
473
464
474
475
476
477
51 unmodified lines
529
530
531
522
523
532
533
534
535
536
28 unmodified lines
565
566
567
558
568
569
570
571
85 unmodified lines
657
658
659
650
660
661
662
663
16 unmodified lines
680
681
682
673
674
683
684
685
686
677
687
688
689
690
6 unmodified lines
697
698
699
690
691
700
701
702
703
704
309 unmodified lines

stats           *statsCollector
    logger          *slog.Logger
    sourceConn      *gitproto.Conn
    targetConn      *gitproto.Conn
    sourceService   *gitproto.RefService
    targetAdv       *packp.AdvRefs
    targetFeatures  gitproto.TargetFeatures
    targetPolicy    planner.RelayTargetPolicy
    targetPusher    gitproto.Pusher
    sourceRefMap    map[plumbing.ReferenceName]plumbing.Hash
    targetRefMap    map[plumbing.ReferenceName]plumbing.Hash
    target          *targetSession
    measurementDone func() Measurement
}

type targetSession struct {
    conn      *gitproto.Conn
    adv       *packp.AdvRefs
    refMap    map[plumbing.ReferenceName]plumbing.Hash
    features  gitproto.TargetFeatures
    policy    planner.RelayTargetPolicy
    pusher    gitproto.Pusher
}

// newSession performs the shared setup: protocol validation, mapping validation,
// connection creation, and ref discovery.
func newSession(ctx context.Context, cfg Config, needTarget bool) (*syncSession, error) {
31 unmodified lines

s.sourceRefMap = gitproto.RefHashMap(sourceRefs)

if needTarget {
        s.targetConn, err = newConn(cfg.Target, "target", s.stats)
        targetConn, err := newConn(cfg.Target, "target", s.stats)
        if err != nil {
            return nil, fmt.Errorf("create target transport: %w", err)
        }
        s.targetAdv, err = gitproto.AdvertisedRefsV1(ctx, s.targetConn, transport.ReceivePackService)
        targetAdv, err := gitproto.AdvertisedRefsV1(ctx, targetConn, transport.ReceivePackService)
        if err != nil {
            return nil, fmt.Errorf("list target refs: %w", err)
        }
        targetRefSlice, err := gitproto.AdvRefsToSlice(s.targetAdv)
        targetRefSlice, err := gitproto.AdvRefsToSlice(targetAdv)
        if err != nil {
            return nil, fmt.Errorf("decode target refs: %w", err)
        }
        s.targetRefMap = gitproto.RefHashMap(targetRefSlice)
        s.targetFeatures = gitproto.TargetFeaturesFromAdvRefs(s.targetAdv)
        s.targetPolicy = planner.RelayTargetPolicy{
            CapabilitiesKnown: s.targetFeatures.Known,
            NoThin:            s.targetFeatures.NoThin,
        }
targetRefMap := gitproto.RefHashMap(targetRefSlice)
targetFeatures := gitproto.TargetFeaturesFromAdvRefs(targetAdv)
s.target = &targetSession{
        conn:     targetConn,
        adv:      targetAdv,
        refMap:   targetRefMap,
        features: targetFeatures,
        policy: planner.RelayTargetPolicy{
            CapabilitiesKnown: targetFeatures.Known,
            NoThin:            targetFeatures.NoThin,
        },
        pusher: gitproto.NewPusher(targetConn, targetAdv, cfg.Verbose),
    }
    s.targetPusher = gitproto.NewPusher(s.targetConn, s.targetAdv, cfg.Verbose)
    }

return s, nil
}

11 unmodified lines

stats := s.stats
    sourceService := s.sourceService
    sourceRefMap := s.sourceRefMap
    targetRefMap := s.targetRefMap
    targetRefMap := s.target.refMap

desiredRefs, managedTargets, err := planner.BuildDesiredRefs(sourceRefMap, planConfig(cfg))
    if err != nil {
60 unmodified lines

if !cfg.DryRun && result.Blocked > 0 {
        return result, fmt.Errorf("blocked %d ref update(s); rerun with --force where appropriate", result.Blocked)
    }
    result.RelayReason = planner.RelayFallbackReason(cfg.Force, cfg.Prune, cfg.DryRun, pushPlans, s.targetPolicy)
    result.RelayReason = planner.RelayFallbackReason(cfg.Force, cfg.Prune, cfg.DryRun, pushPlans, s.target.policy)

if !cfg.DryRun {
        // Try incremental relay first
51 unmodified lines

return Result{}, fmt.Errorf("no source refs matched")
    }

_, reason := planner.CanBootstrapRelay(cfg.Force, cfg.Prune, desiredRefs, s.targetRefMap)
    result, err := bootstrapWithInputs(ctx, s, desiredRefs, s.targetRefMap, reason)
    _, reason := planner.CanBootstrapRelay(cfg.Force, cfg.Prune, desiredRefs, s.target.refMap)
    result, err := bootstrapWithInputs(ctx, s, desiredRefs, s.target.refMap, reason)
    result.Measurement = s.measurementDone()
    return result, err
}
28 unmodified lines

if cfg.Target.URL != "" {
        result.TargetURL = cfg.Target.URL
        result.TargetCaps = gitproto.AdvRefsCaps(s.targetAdv)
        result.TargetCaps = gitproto.AdvRefsCaps(s.target.adv)
        result.Stats = s.stats.snapshot()
        result.Measurement = s.measurementDone()
    }
85 unmodified lines

relayReason string,
) (Result, error) {
    bResult, err := bstrap.Execute(ctx, bstrap.Params{
        SourceConn: s.sourceConn, SourceService: s.sourceService, TargetPusher: s.targetPusher,
        SourceConn: s.sourceConn, SourceService: s.sourceService, TargetPusher: s.target.pusher,
        DesiredRefs: desiredRefs, TargetRefs: targetRefs,
        MaxPackBytes: s.cfg.MaxPackBytes, BatchMaxPack: s.cfg.BatchMaxPackBytes,
        Verbose: s.cfg.Verbose, Logger: s.logger,
16 unmodified lines

pushPlans []planner.BranchPlan,
) (incremental.Result, error) {
    return incremental.Execute(ctx, incremental.Params{
        SourceConn: s.sourceConn, SourceService: s.sourceService, TargetPusher: s.targetPusher,
        DesiredRefs: desiredRefs, TargetRefs: s.targetRefMap,
        SourceConn: s.sourceConn, SourceService: s.sourceService, TargetPusher: s.target.pusher,
        DesiredRefs: desiredRefs, TargetRefs: s.target.refMap,
        PushPlans: pushPlans, MaxPackBytes: s.cfg.MaxPackBytes,
        CanRelay: func(force, prune, dryRun bool, plans []planner.BranchPlan) (bool, string) {
            return planner.CanIncrementalRelay(force, prune, dryRun, plans, s.targetPolicy)
            return planner.CanIncrementalRelay(force, prune, dryRun, plans, s.target.policy)
        },
        CanTagRelay: planner.CanFullTagCreateRelay,
    }, planConfig(s.cfg))
6 unmodified lines

pushPlans []planner.BranchPlan,
) error {
    return materialized.Execute(ctx, materialized.Params{
        Store: store, SourceConn: s.sourceConn, SourceService: s.sourceService, TargetPusher: s.targetPusher,
        DesiredRefs: desiredRefs, TargetRefs: s.targetRefMap,
        Store: store, SourceConn: s.sourceConn, SourceService: s.sourceService, TargetPusher: s.target.pusher,
        DesiredRefs: desiredRefs, TargetRefs: s.target.refMap,
        PushPlans: pushPlans, MaxObjects: s.cfg.MaterializedMaxObjects,
    })
}`