novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

D5: indexer_drop_pending refuses a merge no later merge covers, unless abandon (issue 1966, ADR 0633 §5, ADR 0668 §5)

Browse Source 
#650, its repository's head, was dropped on the advice of a line that named a drop where nothing
covered it; its changes would never have been taken in. The rule lived only in the record and the
advice; the tool now checks it, and the advice names the read again.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-10 03:01:38 +01:00
parent 416e270899
commit 7bdc41c0d3
5 changed files with 56 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/takein.go
+2 -2
View File
@@ -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)
modules/mesh-delivery-indexer/internal/index/pending.go
+24 -5
View File
@@ -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
modules/mesh-delivery-indexer/internal/index/pending_test.go
+14 -1
View File
@@ -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 {
modules/mesh-delivery-indexer/internal/index/pending_tools.go
+7 -5
View File
@@ -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.<sequence>, 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.<sequence>, 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 " +
modules/mesh-delivery-indexer/internal/index/takein.go
+9 -3
View File
@@ -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.<take-in|drop>.<key>`, 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
}