Add BestEffort policy and ActionWarn for per-ref reject downgrades · Entire
Add BestEffort policy and ActionWarn for per-ref reject downgrades
0990568→main· Soph·2mo ago·9 files·+207 added/-21 removed
The receive-pack response is per-ref: the server can accept some refs and reject others in the same push (GitHub's hidden-ref refusals are the load-bearing case). Previously sendReceivePack treated any ng as a fatal error via report.Error(); this commit threads an OnRejection callback down through Pusher and the free Push* functions so callers can opt into receiving per-ref ng statuses without short-circuiting the push. Pack-level unpack failures stay fatal.
SyncPolicy.BestEffort wires that callback up at session construction. The session collects rejections in a map; after each strategy returns, applyRejections walks the plans and downgrades matching entries to a new ActionWarn with the server's reason in plan.Reason. Result.Warned counts the downgrades, complementing Pushed/Skipped/Blocked/Deleted, and the human-readable summary line surfaces it.
The integration test wires a target receive-pack hook that returns "deny updating a hidden ref" for refs/notes/commits while accepting the branch ref, runs with AllRefs+BestEffort, and asserts the notes plan ends up as ActionWarn with the reason carried through.
Co-Authored-By: Claude Opus 4.7 (1M context) noreply@anthropic.com
Sessions
e5feb8b3d187View transcript
Changes
Mclient.go+1
internal
gitproto
Mpush.go+42/-13
Mpush_test.go+5/-5
planner
Mtypes.go+6
syncer
Mintegration_test.go+75
Msyncer.go+65/-3
internalbridge
Mconfig.go+2
Mmodel.go+3
Mtypes.go+8
229 unmodified lines
230
231
232
233
234
235
236
229 unmodified lines
IncludeTags: policy.IncludeTags,
Force: policy.Force,
Prune: policy.Prune,
BestEffort: policy.BestEffort,
Protocol: internalbridge.ProtocolMode(policy.Protocol),
}
}
Mclient.go+1
25 unmodified lines
26
27
28
29
30
31
32
33
34
35
30
31
32
36
37
38
39
40
41
42
3 unmodified lines
46
47
48
42
49
50
51
52
53
47
54
55
56
57
58
52
59
60
61
62
41 unmodified lines
104
105
106
100
107
108
109
110
111
112
113
114
115
116
117
118
119
120
28 unmodified lines
149
150
151
141
142
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
8 unmodified lines
180
181
182
183
184
185
186
187
188
189
163
190
191
192
193
9 unmodified lines
203
204
205
179
206
207
208
209
10 unmodified lines
220
221
222
223
224
225
226
8 unmodified lines
235
236
237
210
238
239
240
241
11 unmodified lines
253
254
255
256
257
258
259
260
261
233
262
263
264
265
25 unmodified lines
}
// Pusher wraps target-side receive-pack state behind a smaller execution API.
// OnRejection, when non-nil, is invoked with each per-ref rejection reported
// by the target's receive-pack instead of returning a fatal error. The pack
// itself unpacking is still fatal; only individual ref ng statuses are
// downgraded to callbacks. This is how best-effort all-refs mode keeps a
// sync going past hidden-ref refusals (refs/pull/* on GitHub, etc.).
type Pusher struct {
Conn *Conn
Adv *packp.AdvRefs
Verbose bool
Conn *Conn
Adv *packp.AdvRefs
Verbose bool
OnRejection func(refName plumbing.ReferenceName, status string)
}
// NewPusher builds a target-side push executor.
3 unmodified lines
// 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)
return PushPack(ctx, p.Conn, p.Adv, commands, pack, p.Verbose, p.OnRejection)
}
// PushCommands sends ref-only updates without a pack.
func (p Pusher) PushCommands(ctx context.Context, commands []PushCommand) error {
return PushCommands(ctx, p.Conn, p.Adv, commands, p.Verbose)
return PushCommands(ctx, p.Conn, p.Adv, commands, 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)
return PushObjects(ctx, p.Conn, p.Adv, commands, store, hashes, p.Verbose, p.OnRejection)
}
// buildUpdateRequest builds the receive-pack update request.
41 unmodified lines
return req, hasDelete, hasUpdates, nil
}
// sendReceivePack encodes and POSTs a receive-pack request, then decodes the report.
// sendReceivePack encodes and POSTs a receive-pack request, then decodes the
// report. When onRejection is non-nil, per-ref ng statuses are reported via
// the callback instead of erroring; the entire push only fails on transport
// errors or unpack failure.
func sendReceivePack(
ctx context.Context,
conn *Conn,
req *packp.UpdateRequests,
packData io.Reader,
verbose bool,
onRejection func(plumbing.ReferenceName, string),
) error {
var header bytes.Buffer
if err := req.Encode(&header); err != nil {
28 unmodified lines
if err := report.Decode(respReader); err != nil {
return fmt.Errorf("decode report-status: %w", err)
}
if err := report.Error(); err != nil {
return fmt.Errorf("report-status: %w", err)
if onRejection == nil {
if err := report.Error(); err != nil {
return fmt.Errorf("report-status: %w", err)
}
return nil
}
// Best-effort: unpack failure is still fatal (the whole pack went
// nowhere), but per-ref ng statuses go to the callback so the
// caller can downgrade them to warnings.
if report.UnpackStatus != "" && report.UnpackStatus != "ok" {
return fmt.Errorf("report-status: unpack error: %s", report.UnpackStatus)
}
for _, cs := range report.CommandStatuses {
if cs.Status == "" || cs.Status == "ok" {
continue
}
onRejection(cs.ReferenceName, cs.Status)
}
}
return nil
8 unmodified lines
store storer.Storer,
hashes []plumbing.Hash,
verbose bool,
onRejection func(plumbing.ReferenceName, string),
) error {
req, _, hasUpdates, err := buildUpdateRequest(adv, commands, verbose)
if err != nil {
return err
}
if !hasUpdates {
return sendReceivePack(ctx, conn, req, nil, verbose)
return sendReceivePack(ctx, conn, req, nil, verbose, onRejection)
}
useRefDeltas := !adv.Capabilities.Supports(capability.OFSDelta)
9 unmodified lines
done <- pw.Close()
}
err = sendReceivePack(ctx, conn, req, pr, verbose)
err = sendReceivePack(ctx, conn, req, pr, verbose, onRejection)
_ = pr.Close()
encodeErr := <-done
if err != nil {
return err
}
commands []PushCommand,
pack io.ReadCloser,
verbose bool,
onRejection func(plumbing.ReferenceName, string),
) error {
for _, cmd := range commands {
if cmd.Delete {
8 unmodified lines
return err
}
err = sendReceivePack(ctx, conn, req, pack, verbose)
err = sendReceivePack(ctx, conn, req, pack, verbose, onRejection)
closeErr := pack.Close()
if err != nil {
return err
11 unmodified lines
adv *packp.AdvRefs,
commands []PushCommand,
verbose bool,
onRejection func(plumbing.ReferenceName, string),
) error {
req, _, _, err := buildUpdateRequest(adv, commands, verbose)
if err != nil {
return err
}
return sendReceivePack(ctx, conn, req, nil, verbose)
return sendReceivePack(ctx, conn, req, nil, verbose, onRejection)
}
func progressWriter(verbose bool, dest io.Writer) io.Writer {
Minternal/gitproto/push.go+42/-13
146 unmodified lines
147
148
149
150
150
151
152
153
21 unmodified lines
175
176
177
178
178
179
180
181
24 unmodified lines
206
207
208
209
209
210
211
212
45 unmodified lines
258
259
260
261
261
262
263
264
65 unmodified lines
330
331
332
333
333
334
335
336
146 unmodified lines
err := PushPack(context.Background(), conn, adv, []PushCommand{{
Name: "refs/heads/main",
New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
}}, pack, false)
}}, pack, false, nil)
if err != nil {
t.Fatalf("PushPack returned error: %v", err)
}
21 unmodified lines
err := PushPack(context.Background(), conn, adv, []PushCommand{{
Name: "refs/heads/main",
New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
}}, pack, false)
}}, pack, false, nil)
if err == nil {
t.Fatal("expected PushPack to return an error")
}
24 unmodified lines
done <- PushPack(ctx, conn, adv, []PushCommand{{
Name: "refs/heads/main",
New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
}}, pack, false)
}}, pack, false, nil)
}
select {
45 unmodified lines
done <- PushPack(context.Background(), conn, adv, []PushCommand{{
Name: "refs/heads/main",
New: plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
}}, pack, false)
}}, pack, false, nil)
}
select {
65 unmodified lines
err = PushPack(context.Background(), conn, adv, []PushCommand{
{Name: "refs/heads/old", Delete: true},
}, pack, false)
}, pack, false, nil)
if err == nil {
t.Fatal("expected error for delete in pack push")
}
}
Minternal/gitproto/push_test.go+5/-5
31 unmodified lines
32
33
34
35
36
37
38
39
40
41
42
43
31 unmodified lines
ActionDelete Action = "delete"
ActionSkip Action = "skip"
ActionBlock Action = "block"
// ActionWarn is set after a push when the target rejected an
// individual ref update under best-effort policy. The push itself
// succeeded for other refs; this ref carries the server's reason in
// BranchPlan.Reason. Used so AllRefs syncs into hostile targets
// (e.g. GitHub refs/pull/* hidden refs) don't fail the whole run.
ActionWarn Action = "warn"
// RefMapping is a user-specified source:target mapping.
Minternal/planner/types.go+6
2828 unmodified lines
2829
2830
2831
2832
2833
2834
2835
2836
2837
2838
2839
2840
2841
2842
2843
2844
2845
2846
2847
2848
2849
2850
2851
2852
2853
2854
2855
2856
2857
2858
2859
2860
2861
2862
2863
2864
2865
2866
2867
2868
2869
2870
2871
2872
2873
2874
2875
2876
2877
2878
2879
2880
2881
2882
2883
2884
2885
2886
2887
2888
2889
2890
2891
2892
2893
2894
2895
2896
2897
2898
2899
2900
2901
2902
2903
2904
2905
2906
2907
2908
2909
2828 unmodified lines
assertHeadsMatch(t, sourceRepo, targetRepo, testBranch)
}
// TestRun_IntegrationAllRefsBestEffortDowngradesNgToWarn verifies that with
// BestEffort the per-ref ng status returned by the target receive-pack is
// downgraded to ActionWarn instead of failing the whole sync — the mode that
// makes AllRefs usable against hosts with hidden refs (GitHub refs/pull/*).
func TestRun_IntegrationAllRefsBestEffortDowngradesNgToWarn(t *testing.T) {
sourceRepo, sourceFS := newSourceRepo(t)
makeCommits(t, sourceRepo, sourceFS, 2)
head, err := sourceRepo.Reference(plumbing.NewBranchReferenceName(testBranch), true)
if err != nil {
t.Fatalf("resolve source head: %v", err)
}
notesRef := plumbing.ReferenceName("refs/notes/commits")
if err := sourceRepo.Storer.SetReference(plumbing.NewHashReference(notesRef, head.Hash())); err != nil {
t.Fatalf("set source notes ref: %v", err)
}
targetRepo, err := git.Init(memory.NewStorage())
if err != nil {
t.Fatalf("init target repo: %v", err)
}
sourceServer := newSmartHTTPRepoServerV2(t, sourceRepo)
targetServer := newSmartHTTPRepoServer(t, targetRepo)
defer sourceServer.Close()
defer targetServer.Close()
// Reject the notes ref (mimicking GitHub's "deny updating a hidden ref")
// while accepting the branch ref. Pack itself unpacks fine.
targetServer.receivePackHook = func(req *packp.UpdateRequests, _ bool) *packp.ReportStatus {
report := packp.NewReportStatus()
report.UnpackStatus = "ok"
for _, cmd := range req.Commands {
status := "ok"
if cmd.Name == notesRef {
status = "deny updating a hidden ref"
}
report.CommandStatuses = append(report.CommandStatuses, &packp.CommandStatus{
ReferenceName: cmd.Name,
Status: status,
})
}
return report
}
result, err := Run(context.Background(), Config{
Source: Endpoint{URL: sourceServer.RepoURL()},
Target: Endpoint{URL: targetServer.RepoURL()},
ProtocolMode: protocolModeV2,
AllRefs: true,
BestEffort: true,
})
if err != nil {
t.Fatalf("expected best-effort sync to succeed despite ng: %v", err)
}
if result.Warned != 1 {
t.Fatalf("expected Warned=1, got %d (result: %+v)", result.Warned, result)
}
var foundWarn bool
for _, plan := range result.Plans {
if plan.TargetRef == notesRef {
if plan.Action != ActionWarn {
t.Errorf("expected notes ref Action=warn, got %s", plan.Action)
}
if !strings.Contains(plan.Reason, "deny updating a hidden ref") {
t.Errorf("expected rejection reason in plan.Reason, got %q", plan.Reason)
}
foundWarn = true
}
}
if !foundWarn {
t.Fatal("expected to find the notes ref in result.Plans")
}
}
// TestRun_IntegrationAllRefsRejectsCustomMappingWithoutAllRefs locks in the
// validation gate: mapping a non-branch/non-tag ref errors when AllRefs is
// not set, so the strict default flow stays loud about unsupported refs.
Minternal/syncer/integration_test.go+75
79 unmodified lines
80
81
82
83
84
85
86
21 unmodified lines
108
109
110
111
112
113
114
20 unmodified lines
135
136
137
138
139
140
141
15 unmodified lines
157
158
159
157
158
160
161
162
163
164
220 unmodified lines
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
53 unmodified lines
468
469
470
471
472
473
474
475
476
477
478
106 unmodified lines
585
586
587
588
589
590
591
592
593
594
595
596
153 unmodified lines
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
85 unmodified lines
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
136 unmodified lines
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
961
1023
1024
1025
1026
79 unmodified lines
Mode string
Force bool
Prune bool
BestEffort bool
MaxPackBytes int64
TargetMaxPackBytes int64
MaterializedMaxObjects int
21 unmodified lines
ActionDelete = planner.ActionDelete
ActionSkip = planner.ActionSkip
ActionBlock = planner.ActionBlock
ActionWarn = planner.ActionWarn
)
type RefInfo struct {
20 unmodified lines
Skipped int `json:"skipped"`
Blocked int `json:"blocked"`
Deleted int `json:"deleted"`
Warned int `json:"warned"`
DryRun bool `json:"dryRun"`
OperationMode string `json:"operationMode"`
Relay bool `json:"relay"`
15 unmodified lines
lines = append(lines, planner.FormatPlanLine(plan))
}
summary := fmt.Sprintf(
"summary: pushed=%d deleted=%d skipped=%d blocked=%d mode=%s protocol=%s relay=%t relay-mode=%s relay-reason=%s batching=%t batch-count=%d planned-batches=%d",
r.Pushed, r.Deleted, r.Skipped, r.Blocked, r.OperationMode, r.Protocol, r.Relay, r.RelayMode, r.RelayReason, r.Batching, r.BatchCount, r.PlannedBatchCount,
"summary: pushed=%d deleted=%d skipped=%d blocked=%d warned=%d mode=%s protocol=%s relay=%t relay-mode=%s relay-reason=%s batching=%t batch-count=%d planned-batches=%d",
r.Pushed, r.Deleted, r.Skipped, r.Blocked, r.Warned, r.OperationMode, r.Protocol, r.Relay, r.RelayMode, r.RelayReason, r.Batching, r.BatchCount, r.PlannedBatchCount,
)
if r.DryRun {
summary += " dry-run=true"
220 unmodified lines
return &clone
}
// applyRejections downgrades plans to ActionWarn when their target ref
// appears in the session's collected rejections. Returns the number of
// plans that were downgraded so callers can update Result counts.
func (s *syncSession) applyRejections(plans []BranchPlan) int {
if len(s.rejections) == 0 {
return 0
}
warned := 0
for i := range plans {
status, ok := s.rejections[plans[i].TargetRef]
if !ok {
continue
}
plans[i].Action = ActionWarn
if status == "" {
plans[i].Reason = "target rejected ref update"
} else {
plans[i].Reason = "target rejected ref update: " + status
}
warned++
}
return warned
}
func planConfig(cfg Config) planner.PlanConfig {
return planner.PlanConfig{
Branches: cfg.Branches,
53 unmodified lines
target *targetSession
measurementDone func() Measurement
progress *progressReporter
// rejections collects per-ref ng statuses reported by the target's
// receive-pack when BestEffort is set. The map is the closure backing
// the Pusher.OnRejection callback wired in newSession; consult it
// after a strategy returns to downgrade matching plans to ActionWarn.
rejections map[plumbing.ReferenceName]string
}
// finish releases any resources owned by the session — currently the live
106 unmodified lines
},
pusher: gitproto.NewPusher(targetConn, targetAdv, cfg.Verbose),
}
if cfg.BestEffort {
s.rejections = make(map[plumbing.ReferenceName]string)
s.target.pusher.OnRejection = func(name plumbing.ReferenceName, status string) {
s.rejections[name] = status
}
}
}
// Start the live progress ticker only after auth resolution and the
153 unmodified lines
}
if !s.cfg.DryRun {
warned := s.applyRejections(pushPlans)
if warned > 0 {
s.applyRejections(result.Plans)
result.Warned += warned
}
}
for _, plan := range pushPlans {
switch plan.Action {
case ActionCreate, ActionUpdate:
result.Pushed++
case ActionDelete:
result.Deleted++
case ActionWarn:
// already counted via result.Warned
case ActionSkip, ActionBlock:
// not applicable in this context
}
85 unmodified lines
result.RelayReason = repResult.RelayReason
}
if !s.cfg.DryRun {
warned := s.applyRejections(pushPlans)
if warned > 0 {
s.applyRejections(result.Plans)
result.Warned += warned
}
}
for _, plan := range pushPlans {
switch plan.Action {
case ActionCreate, ActionUpdate:
result.Pushed++
case ActionDelete:
result.Deleted++
case ActionWarn:
// already counted via result.Warned
case ActionSkip, ActionBlock:
// not applicable in this context
}
136 unmodified lines
if err != nil {
return Result{}, fmt.Errorf("bootstrap execute: %w", err)
}
plans := bResult.Plans
warned := s.applyRejections(plans)
pushed := bResult.Pushed - warned
if pushed < 0 {
pushed = 0
}
return Result{
Plans: bResult.Plans, Pushed: bResult.Pushed, OperationMode: s.cfg.Mode,
Plans: plans, Pushed: pushed, Warned: warned, OperationMode: s.cfg.Mode,
Relay: bResult.Relay, RelayMode: bResult.RelayMode, RelayReason: bResult.RelayReason,
Batching: bResult.Batching, BatchCount: bResult.BatchCount,
PlannedBatchCount: bResult.PlannedBatchCount, TempRefs: bResult.TempRefs,