Merge pull request #94 from entireio/fix/batch-ref-updates-receive-pack · Entire

Merge pull request #94 from entireio/fix/batch-ref-updates-receive-pack

0529f86→main·

Soph·4w ago·6 files·+402 added/-30 removed

Batch ref-update commands to stay under receive-pack cap

Changes

6

78 unmodified lines

79
80
81
82
83
84
85

78 unmodified lines

cmd.Flags().BoolVar(&jsonOutput, "json", false, "print JSON output")
 cmd.Flags().Int64Var(&req.Options.MaxPackBytes, "max-pack-bytes", 0, "abort bootstrap if the streamed source pack exceeds this many bytes")
 cmd.Flags().Int64Var(&req.Options.TargetMaxPackBytes, "target-max-pack-bytes", 0, "target receive-pack body size limit; batches are planned and auto-subdivided to fit")
 cmd.Flags().IntVar(&req.Options.TargetMaxRefUpdates, "target-max-ref-updates", 0, "max ref-update commands per receive-pack request; 0 uses the default (env GITSYNC_MAX_REF_UPDATES_PER_PUSH or 5000). Raise for entire-server targets (up to 25000); lower for providers that reject large ref pushes")
 cmd.Flags().StringVar(&req.Options.BootstrapStrategy, "bootstrap-strategy", "", "checkpoint chain ordering: \"first-parent\" (default) or \"topo\". Use \"topo\" for merge-heavy repos where individual first-parent steps drag in unboundedly large side branches; requires the target to allow non-fast-forward updates on the refs/gitsync/ namespace")
 addProtocolFlag(cmd, &protocolVal)
 cmd.Flags().BoolVarP(&req.Options.Verbose, "verbose", "v", false, "verbose logging")

Mcmd/git-sync/bootstrap.go+1

137 unmodified lines

138
139
140
141
142
143
144

137 unmodified lines

cmd.Flags().IntVar(&req.Options.MaterializedMaxObjects, "materialized-max-objects", unstable.DefaultMaterializedMaxObjects, "abort non-relay materialized syncs above this many objects")
 cmd.Flags().Int64Var(&req.Options.MaxPackBytes, "max-pack-bytes", 0, "abort bootstrap-relay push if the streamed source pack exceeds this many bytes")
 cmd.Flags().Int64Var(&req.Options.TargetMaxPackBytes, "target-max-pack-bytes", 0, "target receive-pack body size limit; batches are planned and auto-subdivided to fit")
 cmd.Flags().IntVar(&req.Options.TargetMaxRefUpdates, "target-max-ref-updates", 0, "max ref-update commands per receive-pack request; 0 uses the default (env GITSYNC_MAX_REF_UPDATES_PER_PUSH or 5000). Raise for entire-server targets (up to 25000); lower for providers that reject large ref pushes")
 cmd.Flags().StringVar(&req.Options.BootstrapStrategy, "bootstrap-strategy", "", "checkpoint chain ordering for bootstrap: \"first-parent\" (default) or \"topo\". Use \"topo\" for merge-heavy repos where individual first-parent steps drag in unboundedly large side branches; requires the target to allow non-fast-forward updates on the refs/gitsync/ namespace")
 addProtocolFlag(cmd, &protocolVal)
 cmd.Flags().BoolVarP(&req.Options.Verbose, "verbose", "v", false, "verbose logging")

Mcmd/git-sync/syncplan.go+1

8 unmodified lines

9
10
11
12
13
14
15
28 unmodified lines

44
45
46
47
48
49
50
51
52
53
54
1 unmodified line

56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
55
146
147
148
149
150
151
61
152
153
154
155
156
66
157
158
159
160
288 unmodified lines

449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
2 unmodified lines

509
510
511
369
512
513
514
515
187 unmodified lines

703
704
705
706
707
708
709
4 unmodified lines

714
715
716
573
717
718
719
720
721
722
723
724
725
726
7 unmodified lines

734
735
736
737
738
739
740
741
742
743
8 unmodified lines

752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781

8 unmodified lines

"io"
    "os"
    "slices"
    "strconv"
    "strings"
    "sync/atomic"
    "time"
28 unmodified lines

Adv         *packp.AdvRefs
    Verbose     bool
    OnRejection func(refName plumbing.ReferenceName, status string)

// MaxRefUpdates caps ref-update commands per receive-pack request. Zero
    // uses the env-or-default limit (see MaxRefUpdatesEnv); a positive value
    // overrides it — e.g. from the --target-max-ref-updates flag.
    MaxRefUpdates int
}

// NewPusher builds a target-side push executor.
1 unmodified line

return &Pusher{Conn: conn, Adv: adv, Verbose: verbose}

// defaultMaxRefUpdatesPerPush bounds how many ref-update commands ride in a
// single receive-pack request. The default is deliberately conservative:
// GitHub returns 500 Internal Server Error when a single push updates ~10k refs
// at once but accepts 5k, so 5_000 mirrors a many-ref repo there without
// tripping its (undocumented) ceiling. entire-server tolerates far more — its
// hard cap is 25_000 (server/githttp.maxRefUpdateCommands) — so trusted callers
// pushing to entire-server raise this via MaxRefUpdatesEnv to cut round trips.
//
// Splitting is safe because the pack accompanying the first batch carries every
// object for the whole push: receive-pack commits the entire received pack into
// the object store (entire-server via CommitQuarantinedFanout, canonical git via
// tmp_objdir_migrate — neither prunes objects unreachable from the pushed tips),
// so the remaining batches only need to move ref pointers and carry no pack.
const defaultMaxRefUpdatesPerPush = 5_000

// MaxRefUpdatesEnv overrides defaultMaxRefUpdatesPerPush with a positive
// integer. Raise it for targets known to accept large ref-update pushes (e.g.
// entire-server, up to its 25_000 cap) to reduce round trips; lower it for a
// provider that rejects even the default. Invalid or non-positive values fall
// back to the default.
const MaxRefUpdatesEnv = "GITSYNC_MAX_REF_UPDATES_PER_PUSH"

// maxRefUpdatesPerPush is resolved once from the environment so the limit can be
// tuned per target without rebuilding (see MaxRefUpdatesEnv).
var maxRefUpdatesPerPush = resolveMaxRefUpdatesPerPush()

func resolveMaxRefUpdatesPerPush() int {
    if v := os.Getenv(MaxRefUpdatesEnv); v != "" {
        if n, err := strconv.Atoi(v); err == nil && n > 0 {
            return n
        }
    }
    return defaultMaxRefUpdatesPerPush
}

// effectiveMaxRefUpdates resolves a per-push limit: a positive override wins,
// otherwise the env-or-default limit applies.
func effectiveMaxRefUpdates(maxRefUpdates int) int {
    if maxRefUpdates > 0 {
        return maxRefUpdates
    }
    return maxRefUpdatesPerPush
}

// chunkRefUpdates splits commands into batches no larger than limit. Input that
// already fits is returned as a single batch (including the empty slice, so
// callers preserve their one-request behavior).
func chunkRefUpdates(commands []PushCommand, limit int) [][]PushCommand {
    if len(commands) <= limit {
        return [][]PushCommand{commands}
    }
    
    batches := make([][]PushCommand, 0, (len(commands)+limit-1)/limit)
    for start := 0; start < len(commands); start += limit {
        end := min(start+limit, len(commands))
        batches = append(batches, commands[start:end])
    }
    return batches
}

// splitFirstBatch peels off the first batch (up to limit) so a push can carry
// the pack with that batch and send the remainder as ref-only follow-ups. rest
// is nil when commands already fit in a single request.
func splitFirstBatch(commands []PushCommand, limit int) (first, rest []PushCommand) {
    if len(commands) <= limit {
        return commands, nil
    }
    return commands[:limit], commands[limit:]
}

// logRefUpdateBatch reports completion of one ref-update batch to the progress
// writer. Ref-only follow-up batches push with progress suppressed (their
// sideband carries nothing but a bare "target:" line per batch), so this is the
// only per-batch signal; it stays quiet unless verbose and the push actually
// spanned multiple batches.
func logRefUpdateBatch(conn Conn, verbose bool, batchNum, totalBatches, refs int) {
    if !verbose || totalBatches <= 1 {
        return
    }
    w := conn.ProgressWriter()
    if w == nil {
        w = os.Stderr
    }
    fmt.Fprintf(w, "target: pushed ref-update batch %d/%d (%d refs)\n", batchNum, totalBatches, refs)
}

// PushPack streams a pack to the target.
func (p *Pusher) PushPack(ctx context.Context, commands []PushCommand, pack io.ReadCloser) error {
    return PushPack(ctx, p.Conn, p.Adv, commands, pack, p.Verbose, p.OnRejection)

return PushPack(ctx, p.Conn, p.Adv, commands, pack, p.MaxRefUpdates, p.Verbose, p.OnRejection)
}

// PushCommands sends ref-only updates. Creates/updates carry an empty pack;
// delete-only pushes carry no pack. See the package-level PushCommands.
func (p *Pusher) PushCommands(ctx context.Context, commands []PushCommand) error {
    return PushCommands(ctx, p.Conn, p.Adv, commands, p.Verbose, p.OnRejection)

return PushCommands(ctx, p.Conn, p.Adv, commands, p.MaxRefUpdates, p.Verbose, p.OnRejection)
}

// PushObjects encodes and pushes locally materialized objects.
func (p *Pusher) PushObjects(ctx context.Context, commands []PushCommand, store storer.Storer, hashes []plumbing.Hash) error {
    return PushObjects(ctx, p.Conn, p.Adv, commands, store, hashes, p.Verbose, p.OnRejection)

return PushObjects(ctx, p.Conn, p.Adv, commands, store, hashes, p.MaxRefUpdates, p.Verbose, p.OnRejection)
}

// buildUpdateRequest builds the receive-pack update request.
288 unmodified lines

// PushObjects pushes locally-materialized objects to the target.
//
// A push within the per-request ref-update limit (see effectiveMaxRefUpdates)
// is a single atomic receive-pack request. A larger push is split: the
// materialized pack — which carries every object for the whole push — rides
// with the first batch of object-bearing commands, then the remaining refs (and
// any deletes) move as ref-only updates because the objects are already
// committed.
func PushObjects(
    ctx context.Context,
    conn Conn,
    adv *packp.AdvRefs,
    commands []PushCommand,
    store storer.Storer,
    hashes []plumbing.Hash,
    maxRefUpdates int,
    verbose bool,
    onRejection func(plumbing.ReferenceName, string),
) error {
    limit := effectiveMaxRefUpdates(maxRefUpdates)
    if len(commands) <= limit {
        return pushObjectsBatch(ctx, conn, adv, commands, store, hashes, verbose, onRejection)
    }

updates := make([]PushCommand, 0, len(commands))
    var deletes []PushCommand
    for _, c := range commands {
        if c.Delete {
            deletes = append(deletes, c)
        } else {
            updates = append(updates, c)
        }
    }

if len(updates) > 0 {
        first, rest := splitFirstBatch(updates, limit)
        if err := pushObjectsBatch(ctx, conn, adv, first, store, hashes, verbose, onRejection); err != nil {
            return err
        }
        if len(rest) > 0 {
            if err := PushCommands(ctx, conn, adv, rest, maxRefUpdates, verbose, onRejection); err != nil {
                return err
            }
        }
    }
    if len(deletes) > 0 {
        return PushCommands(ctx, conn, adv, deletes, maxRefUpdates, verbose, onRejection)
    }
    return nil
}

// pushObjectsBatch encodes the selected objects into a pack and sends one
// receive-pack request for commands.
//
// Delta selection runs synchronously up front via
// packfile.DeltaSelector. The selected objects are then handed back to
// a packfile.Encoder behind a passthrough ObjectSelector, so the

// the mid-stream stall that occurs when Encode runs selection itself —
    // CDN edges treat the resulting idle gap as a stalled upload and close
    // the connection. See go-git PR #2142 for the API hook.
func PushObjects(
func pushObjectsBatch(
    ctx context.Context,
    conn Conn,
    adv *packp.AdvRefs,
    commands []PushCommand,
    pack io.ReadCloser,
    maxRefUpdates int,
    verbose bool,
    onRejection func(plumbing.ReferenceName, string),
) error {

if err != nil {
        _ = pack.Close()
        return err
    }

if len(rest) > 0 {
        return PushCommands(ctx, conn, adv, rest, maxRefUpdates, verbose, onRejection)
    }
    return nil
}

}

Minternal/gitproto/push.go+183/-5

11 unmodified lines

12 13 14 15 16 17 18 134 unmodified lines

153 154 155 155 156 157 158 159 20 unmodified lines

180 181 182 182 183 184 185 186 23 unmodified lines

210 211 212 212 213 214 215 216 49 unmodified lines

266 267 268 268 269 270 271 272 50 unmodified lines

323 324 325 325 326 327 328 329 89 unmodified lines

419 420 421 421 422 423 424 425 12 unmodified lines

438 439 440 440 441 442 443 444 50 unmodified lines

495 496 497 497 498 499 500 501 234 unmodified lines

736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924

11 unmodified lines

"net/http" "net/http/httptest" "strings" "sync" "testing" "time"

134 unmodified lines

err := PushPack(context.Background(), conn, adv, []PushCommand{{ Name: "refs/heads/main", New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}, pack, false, nil) }}, pack, 0, false, nil) if err != nil { t.Fatalf("PushPack returned error: %v", err) } }

12 unmodified lines

err := PushPack(context.Background(), conn, adv, []PushCommand{{ Name: "refs/heads/main", New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}, pack, false, nil) }, pack, 0, false, nil) if err == nil { t.Fatal("expected PushPack to return an error") } }

23 unmodified lines

done <- PushPack(ctx, conn, adv, []PushCommand{{ Name: "refs/heads/main", New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}, pack, false, nil) }}, pack, 0, false, nil) })

done <- PushPack(context.Background(), conn, adv, []PushCommand{{ Name: "refs/heads/main", New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), }}, pack, false, nil) }}, pack, 0, false, nil) }) }

select {

)
}).

}

}

}

}`