From 6c79e456a100b75b0ec287c85654ee3965d14c2d Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 27 Oct 2026 00:20:16 +0100 Subject: [PATCH] The indexer confirms a forge message's ack with the server before it fills taken keys (issue 1280) --- .../cmd/mesh-delivery-indexer/main.go | 8 +++++++- .../cmd/mesh-delivery-indexer/main_test.go | 10 ++++++++-- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 24d94ed..4c2a403 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -192,6 +192,7 @@ type delivered interface { Subject() string Data() []byte Ack() error + DoubleAck(ctx context.Context) error Term() error } @@ -212,7 +213,12 @@ func handle(ctx context.Context, m delivered, ix *index.Indexer, logger *log.Log case err != nil: logger.Printf("%s: %s#%d left for redelivery: %v", m.Subject(), f.Repository, f.Number, err) default: - _ = m.Ack() + // Waits for the server, so the consumer's info no longer counts the message as unacknowledged when the fill that + // follows asks whether the stream is applied (issue 1280); a failed ack is redelivered after the ack wait, which + // kicks the fill again. + if err := m.DoubleAck(ctx); err != nil { + logger.Printf("%s: %s#%d applied, its acknowledgement was not confirmed: %v", m.Subject(), f.Repository, f.Number, err) + } } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index 19b3658..e49c636 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -16,12 +16,18 @@ type fakeMsg struct { subject string data []byte acked, termd int + doubled int // acked by DoubleAck, which waits for the server } func (f *fakeMsg) Subject() string { return f.subject } func (f *fakeMsg) Data() []byte { return f.data } func (f *fakeMsg) Ack() error { f.acked++; return nil } -func (f *fakeMsg) Term() error { f.termd++; return nil } +func (f *fakeMsg) DoubleAck(context.Context) error { + f.acked++ + f.doubled++ + return nil +} +func (f *fakeMsg) Term() error { f.termd++; return nil } type memStore struct{ vals map[string][]byte } @@ -53,7 +59,7 @@ func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part) m := &fakeMsg{subject: "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull", data: data} handle(context.Background(), m, ix, log.New(&bytes.Buffer{}, "", 0)) c, err := delivery.DecodeCandidate(st.vals["feat/x"]) - if err != nil || len(c.Pulls) != 1 || c.Pulls[0].Modules[0] != "a" || m.acked != 1 || m.termd != 0 { + if err != nil || len(c.Pulls) != 1 || c.Pulls[0].Modules[0] != "a" || m.acked != 1 || m.doubled != 1 || m.termd != 0 { t.Errorf("candidate %+v err %v acked %d terminated %d", c, err, m.acked, m.termd) } }