diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go index 8585099d..af5f2e8a 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go @@ -20,7 +20,7 @@ type takerIn interface { SeatDefinitionsAgain(ctx context.Context, module string, pull int, head string) index.TakenIn HoldAgain(ctx context.Context, module string, release bool, by string) index.TakenIn TakeInPendingByHand(ctx context.Context, key, by, why string) index.TakenIn - DropPending(ctx context.Context, key, by, why string) index.TakenIn + DropPending(ctx context.Context, key, by, why string, abandon bool) index.TakenIn TakeInMerge(ctx context.Context, repository, merge string, seq uint64, by, why string) index.TakenIn DropTryCandidate(ctx context.Context, id, by, why string) index.TakenIn RecoverTriesByHand(ctx context.Context, by, why string) index.TakenIn @@ -55,7 +55,7 @@ func takenIn(ctx context.Context, ix takerIn, kv jetstream.KeyValue, e jetstream asked, r = "hand take-in of a merge", ix.TakeInMerge(ctx, a.MergeRepository, a.Merge, a.Sequence, a.By, a.Why) a.Module = r.Module } else if a.Pending != "" && a.Drop { - asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why) + asked, r = "drop of a pending merge", ix.DropPending(ctx, a.Pending, a.By, a.Why, a.Abandon) a.Module = a.Pending } else if a.Pending != "" { asked, r = "hand take-in of what waits", ix.TakeInPendingByHand(ctx, a.Pending, a.By, a.Why) diff --git a/modules/mesh-delivery-indexer/internal/index/pending.go b/modules/mesh-delivery-indexer/internal/index/pending.go index d92c6956..ab039baa 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending.go +++ b/modules/mesh-delivery-indexer/internal/index/pending.go @@ -760,8 +760,8 @@ func (ix *Indexer) holdLost(ctx context.Context, ready, lost []PendingEntry, rea return nil } for _, g := range lost { - ix.logf("take-in: %s (%s#%d) is not taken in and holds its revision, the forge bridge could not give its message: %s; a newer merge of %s covers it, indexer_drop_pending drops it", - g.Key, g.Forge.Repository, g.Forge.Number, reasons[g.Key], g.Forge.Repository) + ix.logf("take-in: %s (%s#%d) is not taken in and holds its revision, the forge bridge could not give its message: %s; %s reads it again; indexer_drop_pending drops it once a later merge of %s covers it", + g.Key, g.Forge.Repository, g.Forge.Number, reasons[g.Key], TakeInPendingTool, g.Forge.Repository) } if !ix.sayTakeIn(ctx, t) { return nil @@ -1226,7 +1226,7 @@ func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string) } take, lost, reasons := ix.loadNewest(ctx, take) if len(lost) > 0 { - r.Answer = fmt.Sprintf("nothing is taken in: the message of %s is not read (%s); a newer merge of its repository covers it, indexer_drop_pending drops it", lost[0].Key, reasons[lost[0].Key]) + r.Answer = fmt.Sprintf("nothing is taken in: the message of %s is not read (%s); asked again, it is read again; indexer_drop_pending drops it once a later merge of its repository covers it", lost[0].Key, reasons[lost[0].Key]) return r } ix.logf("take-in: a hand take-in of %s (%d merges) asked by %s: %s", key, len(take), by, why) @@ -1245,8 +1245,10 @@ func (ix *Indexer) TakeInPendingByHand(ctx context.Context, key, by, why string) // DropPending is the hand drop (indexer_drop_pending, ADR 0633 §5): the one pending merge under key, abandoned // because a later merge replaced it, is removed and nothing of it is taken in. Its wait is said taken, naming the drop; -// the merges left under that wait are said pending again by the evaluation that follows. -func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn { +// the merges left under that wait are said pending again by the evaluation that follows. A merge that no later pending +// merge and no taken merge of its repository covers is refused unless abandon (issue 1966: #650, its repository's head, +// was dropped on the advice of a line that named a drop where nothing covered it; ADR 0668 §5). +func (ix *Indexer) DropPending(ctx context.Context, key, by, why string, abandon bool) TakenIn { r := TakenIn{Module: key} if ix.Pending == nil { r.Answer = "the pending state is not on the bus: nothing waits" @@ -1266,6 +1268,23 @@ func (ix *Indexer) DropPending(ctx context.Context, key, by, why string) TakenIn return r } e := entries[at] + if !abandon { + covered := slices.ContainsFunc(entries[at+1:], func(o PendingEntry) bool { return o.Forge.Repository == e.Forge.Repository }) + if !covered { + taken, err := ix.takenAlready(ctx, forgemsg.GroupMerge{Repository: e.Forge.Repository, Number: e.Forge.Number, Merge: e.Forge.Merge}) + if err != nil { + r.Answer = fmt.Sprintf("%s is not dropped: whether a taken merge of %s covers it is not read: %v", key, e.Forge.Repository, err) + return r + } + covered = taken + } + if !covered { + r.Answer = fmt.Sprintf("%s is not dropped: no later merge of %s is pending and its taken merge does not cover %s, so its "+ + "changes would never be taken in (ADR 0633 §5); %s reads its message again, or abandon: true drops it all the same", + key, e.Forge.Repository, e.Forge.Merge, TakeInPendingTool) + return r + } + } if err := ix.Pending.Delete(ctx, e.Key, e.Revision); err != nil { r.Answer = fmt.Sprintf("%s.%s is not removed: %v", PendingState, key, err) return r diff --git a/modules/mesh-delivery-indexer/internal/index/pending_test.go b/modules/mesh-delivery-indexer/internal/index/pending_test.go index 0ecaef44..0002e64e 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_test.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_test.go @@ -679,12 +679,18 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) t.Fatalf("indexer_pending = %+v", answer) } b.said = nil - if r := b.ix.DropPending(ctx, takein.PendingKey(2), "lead@shanks", "replaced by m.3"); !r.Stored { + if r := b.ix.DropPending(ctx, takein.PendingKey(2), "lead@shanks", "replaced by m.3", false); !r.Stored { t.Fatalf("drop: %+v", r) } if len(b.said) < 2 || !b.said[0].Dropped || b.said[0].Key != takein.PendingKey(2) || b.said[0].By != "lead@shanks" || b.said[1].Key != takein.PendingKey(3) { t.Fatalf("drop said %+v: want m.2 taken (dropped, by) and m.3 pending", b.said) } + // Issue 1966 (#650), ADR 0633 §5: the newest pending merge of its repository, which no taken merge covers, is not + // dropped; the answer names the read again. abandon drops it all the same (below). + if r := b.ix.DropPending(ctx, takein.PendingKey(3), "lead@shanks", "lost message", false); r.Stored || + !strings.Contains(r.Answer, TakeInPendingTool) || len(b.keys()) != 1 { + t.Fatalf("a drop nothing covers: %+v, pending %v", r, b.keys()) + } b.said = nil if r := b.ix.TakeInPendingByHand(ctx, takein.PendingKey(3), "lead@shanks", "the named indexer can never run"); !r.Stored { t.Fatalf("hand take-in: %+v", r) @@ -692,6 +698,13 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633) if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('d') || len(b.said) != 1 || b.said[0].Kind != takein.TakeInTaken || b.said[0].By != "lead@shanks" { t.Errorf("hand take-in: pending %v, avahi at %s, said %+v", b.keys(), b.commit(t, "avahi"), b.said) } + b.arrive(t, 4, b.merge(23, sha('e'), "tree-avahi-5", "tree-indexer-2")) + if r := b.ix.DropPending(ctx, takein.PendingKey(4), "lead@shanks", "abandoned on the operator's word", true); !r.Stored || len(b.keys()) != 0 { + t.Errorf("an abandoned drop: %+v, pending %v", r, b.keys()) + } + if _, _, err := NewPendingAsk(takein.PendingKey(4), "x", "lead@shanks", false, true, time.Unix(0, 0)); err == nil { + t.Errorf("abandon on a take-in was asked") + } }) t.Run("1585 a merge over the value cap is held by reference, a new indexer with an empty memory takes it in through the forge bridge, and a covered pull request leaves its candidate", func(t *testing.T) { big := func(f forgemsg.Forge) forgemsg.Forge { diff --git a/modules/mesh-delivery-indexer/internal/index/pending_tools.go b/modules/mesh-delivery-indexer/internal/index/pending_tools.go index 189cff8a..4ab76300 100644 --- a/modules/mesh-delivery-indexer/internal/index/pending_tools.go +++ b/modules/mesh-delivery-indexer/internal/index/pending_tools.go @@ -21,7 +21,7 @@ func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (T str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s } hand := func(drop bool) func(map[string]any) (any, error) { return func(args map[string]any) (any, error) { - key, value, err := NewPendingAsk(str(args, "key"), str(args, "why"), stdio.CallerOf(args), drop, now()) + key, value, err := NewPendingAsk(str(args, "key"), str(args, "why"), stdio.CallerOf(args), drop, args["abandon"] == true, now()) if err != nil { return nil, err } @@ -93,11 +93,13 @@ func PendingTools(pending, ownBuild Store, ask func(key string, value []byte) (T Run: hand(false)}, {Name: DropPendingTool, Description: "Drop one pending merge by hand, abandoned because a later merge replaced it (ADR 0633 §5): it is removed from " + - "the pending state and nothing of it is taken in; what waited with it is worked out again. A hand act: who and why are " + - "said in the journal and on the take-in.taken event (dropped).", + "the pending state and nothing of it is taken in; what waited with it is worked out again. A merge no later pending merge " + + "and no taken merge of its repository covers is refused, naming indexer_take_in_pending, which reads its message again; " + + "abandon: true drops it all the same. A hand act: who and why are said in the journal and on the take-in.taken event (dropped).", Input: map[string]any{"type": "object", "required": []string{"key", "why"}, "properties": map[string]any{ - "key": map[string]any{"type": "string", "description": "the pending merge's key, m., as indexer_pending names its merges' wait"}, - "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}}}, + "key": map[string]any{"type": "string", "description": "the pending merge's key, m., as indexer_pending names its merges' wait"}, + "why": map[string]any{"type": "string", "description": "why it is dropped, in plain words"}, + "abandon": map[string]any{"type": "boolean", "description": "drop it though no later merge covers it; its changes are then never taken in"}}}, Run: hand(true)}, {Name: TakeInMergeTool, Description: "Take one merge in by hand (ADR 0633 §5, issue 1585): the indexer asks the forge bridge for the merge's merged message " + diff --git a/modules/mesh-delivery-indexer/internal/index/takein.go b/modules/mesh-delivery-indexer/internal/index/takein.go index b20c7006..89505157 100644 --- a/modules/mesh-delivery-indexer/internal/index/takein.go +++ b/modules/mesh-delivery-indexer/internal/index/takein.go @@ -49,6 +49,9 @@ type TakeInAsk struct { // Pending asks the wait under this take-in key to be taken in by hand (indexer_take_in_pending), or with Drop the // one pending merge under it to be dropped (indexer_drop_pending), instead of a module (pending.go, ADR 0633 §5). Pending string `json:"pending,omitempty"` + // Abandon, with Drop, drops a pending merge no later merge covers (ADR 0633 §5 asks a later merge; issue 1966, ADR + // 0668 §5): the hand act says why it is abandoned all the same. + Abandon bool `json:"abandon,omitempty"` // Merge asks the merge of MergeRepository at this commit to be taken in by hand (indexer_take_in_merge): its merged message // is asked of the forge bridge and run through the take-in as a delivered one, under its stream sequence Sequence. Merge string `json:"merge,omitempty"` @@ -137,8 +140,8 @@ func NewMergeAsk(repository, merge string, sequence uint64, why, by string, at t // NewPendingAsk is the value indexer_take_in_pending (drop false) or indexer_drop_pending (drop true) writes: its key // in takeins, `pending..`, and the value. key is a wait's key as indexer_pending names it, for a -// drop one pending merge's. -func NewPendingAsk(key, why, by string, drop bool, at time.Time) (string, []byte, error) { +// drop one pending merge's; abandon drops one no later merge covers. +func NewPendingAsk(key, why, by string, drop, abandon bool, at time.Time) (string, []byte, error) { key = strings.TrimSpace(key) if !strings.HasPrefix(key, "m.") && !(strings.HasPrefix(key, "g.") && !drop) { if drop { @@ -153,6 +156,9 @@ func NewPendingAsk(key, why, by string, drop bool, at time.Time) (string, []byte if drop { act = "drop" } - data, err := json.Marshal(TakeInAsk{V: 1, Pending: key, Drop: drop, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) + if abandon && !drop { + return "", nil, errors.New("abandon is a drop's: indexer_take_in_pending takes a wait in") + } + data, err := json.Marshal(TakeInAsk{V: 1, Pending: key, Drop: drop, Abandon: abandon, Why: strings.TrimSpace(why), By: by, At: at.UTC()}) return "pending." + act + "." + key, data, err }