The indexer confirms a forge message's ack with the server before it fills taken keys (issue 1280)

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)
+		}
 	}
 }
 

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)
 	}
 }