Encapsulate materialized execution stages

f7ab510→main·

Soph·3mo ago·2 files·+50 added/-31 removed

Sessions

de77f2de9b81View transcript

Changes

2

• docs

  • Mrewrite-issue-list.md+1

• internal/strategy/materialized

  • Mmaterialized.go+49/-31

294 unmodified lines

295
296
297
298
299
300
301

294 unmodified lines

- The strategies now also depend on a narrower target-side push executor instead of raw target transport state, and direct strategy tests exercise those boundaries.
- Incremental relay policy decisions are now injected consistently instead of splitting between one injected check and one hard-coded planner call.
- Bootstrap checkpoint planning now carries its graph, probe cache, and prefetched-pack state inside a dedicated internal planner object instead of one large helper function.
- The materialized fallback now runs through an explicit executor with separate stages for tag prefetch, object-closure collection, limit enforcement, and push execution.
- Some helpers still carry broad parameter structs, so this remains partial rather than fully complete.

## Performance And Scalability

Mdocs/rewrite-issue-list.md+1

36 unmodified lines

37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
47
48
54
55
56
57
51
52
53
54
55
56
57
58
58
59
60
61
62
63
64
65
66
67
68
69
62
63
64
71
72
73
74
75
76
77
78
79
65
66
67
82
68
69
84
70
71
72
73
88
74
75
76
77
3 unmodified lines

81
82
83
98
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125

36 unmodified lines

// Fail early rather than OOM (issue #15).
const DefaultMaxMaterializedObjects = 500_000

type executor struct {
    ctx    context.Context
    params Params
}

// Execute runs the materialized fallback: ensures tag objects are local,
// computes the object closure, and pushes to the target.
func Execute(ctx context.Context, p Params) error {
    if len(p.PushPlans) == 0 {
        return nil
    }
    return (&executor{ctx: ctx, params: p}).run()
}

// Ensure tag objects are fetched locally
if err := ensureTagObjects(ctx, p); err != nil {
func (e *executor) run() error {
    if err := e.ensureTagObjects(); err != nil {
        return fmt.Errorf("prepare local objects for push: %w", err)
    }

objects := make([]plumbing.Hash, 0, len(p.PushPlans))
    for _, plan := range p.PushPlans {
        if plan.Action == planner.ActionCreate || plan.Action == planner.ActionUpdate {
            objects = append(objects, plan.SourceHash)
        }
    }
    hashes, err := planner.ObjectsToPush(p.Store, objects, p.TargetRefs);
hashes, err := e.collectObjectClosure()
    if err != nil {
        return fmt.Errorf("compute objects to push: %w", err)
    }

// Issue #15: guard against unbounded memory usage on large non-relay syncs.
    maxObjects := effectiveMaxObjects(p.MaxObjects)
    if len(hashes) > maxObjects {
        return fmt.Errorf(
            "materialized push requires %d objects (limit %d); use bootstrap for large initial syncs",
            len(hashes), maxObjects,
        )
    }
    if err := e.enforceObjectLimit(hashes); err != nil {
        return err
    }

cmds := convert.PlansToPushCommands(p.PushPlans)
    if p.TargetPusher == nil {
        return fmt.Errorf("materialized strategy requires TargetPusher")
    }
    if err := p.TargetPusher.PushObjects(ctx, cmds, p.Store, hashes); err != nil {
        return fmt.Errorf("push target refs: %w", err)
    }
    return nil
    return e.push(hashes)
}

func ensureTagObjects(ctx context.Context, p Params) error {
func (e *executor) ensureTagObjects() error {
    tagDesired := make(map[plumbing.ReferenceName]gitproto.DesiredRef)
    for _, plan := range p.PushPlans {
    for _, plan := range e.params.PushPlans {
        if plan.Kind != planner.RefKindTag {
            continue
        }
        if d, ok := p.DesiredRefs[plan.TargetRef]; ok {
        if d, ok := e.params.DesiredRefs[plan.TargetRef]; ok {
            tagDesired[plan.TargetRef] = gitproto.DesiredRef{
                SourceRef: d.SourceRef, TargetRef: d.TargetRef,
                SourceHash: d.SourceHash, IsTag: true,
    }

if len(tagDesired) == 0 {
        return nil
    }
    err := p.SourceService.FetchToStore(ctx, p.Store, p.SourceConn, tagDesired, nil)
    err := e.params.SourceService.FetchToStore(e.ctx, e.params.Store, e.params.SourceConn, tagDesired, nil)
    if err != nil && err != git.NoErrAlreadyUpToDate {
        return err
    }
    return nil
}

func (e *executor) collectObjectClosure() ([]plumbing.Hash, error) {
    objects := make([]plumbing.Hash, 0, len(e.params.PushPlans))
    for _, plan := range e.params.PushPlans {
        if plan.Action == planner.ActionCreate || plan.Action == planner.ActionUpdate {
            objects = append(objects, plan.SourceHash)
        }
    }
    return planner.ObjectsToPush(e.params.Store, objects, e.params.TargetRefs)
}

func (e *executor) enforceObjectLimit(hashes []plumbing.Hash) error {
    maxObjects := effectiveMaxObjects(e.params.MaxObjects)
    if len(hashes) <= maxObjects {
        return nil
    }
    return fmt.Errorf(
        "materialized push requires %d objects (limit %d); use bootstrap for large initial syncs",
        len(hashes), maxObjects,
    )
}

func (e *executor) push(hashes []plumbing.Hash) error {
    cmds := convert.PlansToPushCommands(e.params.PushPlans)
    if e.params.TargetPusher == nil {
        return fmt.Errorf("materialized strategy requires TargetPusher")
    }
    if err := e.params.TargetPusher.PushObjects(e.ctx, cmds, e.params.Store, hashes); err != nil {
        return fmt.Errorf("push target refs: %w", err)
    }
    return nil
}

func effectiveMaxObjects(limit int) int {
    if limit > 0 {
        return limit
    }