Fix races in ISO message confirmations

broker/events/eventbus.go

@@ -25,6 +25,8 @@ type EventBus interface {
 	Start(ctx common.ExtendedContext) error
 	CreateTask(id string, eventName EventName, data EventData, eventDomain EventDomain, parentId *string, target SignalTarget) (string, error)
 	CreateNotice(id string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, target SignalTarget) (string, error)
+	// CreateNoticeWithID publishes a notice using a caller-allocated event ID.
+	CreateNoticeWithID(eventID string, id string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, target SignalTarget) (string, error)
 	// CreateNoticeWithParent creates a notice linked to a parent event.
 	CreateNoticeWithParent(id string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, parentId *string, target SignalTarget) (string, error)
 	// BeginTask marks a task as processing and emits SignalTaskBegin to the selected target.
@@ -224,15 +226,19 @@ func (p *PostgresEventBus) CreateTask(classId string, eventName EventName, data
 }
 
 func (p *PostgresEventBus) CreateNotice(classId string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, target SignalTarget) (string, error) {
-	return p.createNotice(classId, eventName, data, status, eventDomain, nil, target)
+	return p.CreateNoticeWithID(uuid.NewString(), classId, eventName, data, status, eventDomain, target)
 }
 
 func (p *PostgresEventBus) CreateNoticeWithParent(classId string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, parentId *string, target SignalTarget) (string, error) {
-	return p.createNotice(classId, eventName, data, status, eventDomain, parentId, target)
+	return p.createNotice(uuid.NewString(), classId, eventName, data, status, eventDomain, parentId, target)
 }
 
-func (p *PostgresEventBus) createNotice(classId string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, parentId *string, target SignalTarget) (string, error) {
-	id := uuid.New().String()
+// CreateNoticeWithID publishes a notice using a caller-allocated event ID.
+func (p *PostgresEventBus) CreateNoticeWithID(eventID string, classId string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, target SignalTarget) (string, error) {
+	return p.createNotice(eventID, classId, eventName, data, status, eventDomain, nil, target)
+}
+
+func (p *PostgresEventBus) createNotice(id string, classId string, eventName EventName, data EventData, status EventStatus, eventDomain EventDomain, parentId *string, target SignalTarget) (string, error) {
 	illTransactionID, patronRequestID := getIllTransactionAndPatronRequestId(classId, eventDomain)
 	return id, p.repo.WithTxFunc(p.ctx, func(eventRepo EventRepo) error {
 		event, err := eventRepo.SaveEvent(p.ctx, SaveEventParams{

broker/events/eventbus_test.go

@@ -97,3 +97,40 @@ func (r *exclusiveCheckErrorRepo) GetLatestRequestEventByAction(ctx common.Exten
 func (r *exclusiveCheckErrorRepo) GetPatronRequestEvents(ctx common.ExtendedContext, id string) ([]Event, error) {
 	return nil, nil
 }
+
+func TestCreateNoticeWithID(t *testing.T) {
+	repo := &noticeIDRepo{}
+	bus := NewPostgresEventBus(repo, "")
+	bus.ctx = common.CreateExtCtxWithArgs(context.Background(), nil)
+	id, err := bus.CreateNoticeWithID("allocated-event", "transaction", EventNameRequesterMsgReceived, EventData{}, EventStatusSuccess, EventDomainIllTransaction, SignalConsumers)
+	assert.NoError(t, err)
+	assert.Equal(t, "allocated-event", id)
+	assert.Equal(t, id, repo.saved.ID)
+	assert.Equal(t, "transaction", repo.saved.IllTransactionID)
+	assert.Equal(t, EventTypeNotice, repo.saved.EventType)
+	assert.Equal(t, EventNameRequesterMsgReceived, repo.saved.EventName)
+	assert.Equal(t, EventStatusSuccess, repo.saved.EventStatus)
+	assert.Equal(t, id, repo.notifiedID)
+	assert.Equal(t, SignalNoticeCreated, repo.signal)
+	assert.Equal(t, SignalConsumers, repo.target)
+}
+
+type noticeIDRepo struct {
+	EventRepo
+	saved      Event
+	notifiedID string
+	signal     Signal
+	target     SignalTarget
+}
+
+func (r *noticeIDRepo) WithTxFunc(_ common.ExtendedContext, fn func(EventRepo) error) error {
+	return fn(r)
+}
+func (r *noticeIDRepo) SaveEvent(_ common.ExtendedContext, params SaveEventParams) (Event, error) {
+	r.saved = Event(params)
+	return r.saved, nil
+}
+func (r *noticeIDRepo) Notify(_ common.ExtendedContext, id string, signal Signal, target SignalTarget) error {
+	r.notifiedID, r.signal, r.target = id, signal, target
+	return nil
+}

broker/handler/iso18626-handler.go

@@ -83,7 +83,7 @@ var ErrInvalidStatus = errors.New(string(InvalidStatus))
 var ErrInvalidReason = errors.New(string(InvalidReason))
 var ErrDuplicateRequest = errors.New(string(ReqIsDuplicate))
 
-var waitingReqs = map[string]RequestWait{}
+var waitingReqs = requestWaitRegistry{requests: make(map[string]RequestWait)}
 
 // RequestOptions controls local request processing and is never read from ISO messages.
 type RequestOptions struct {
@@ -570,18 +570,9 @@ func handleRequestingAgencyMessage(ctx common.ExtendedContext, illMessage *iso18
 		return
 	}
 
-	eventId, err := createNotice(ctx, eventBus, illTrans.ID, events.EventNameRequesterMsgReceived, eventData, events.EventStatusSuccess)
-	if err != nil {
+	if err := waitForMessageConfirmation(ctx, eventBus, illTrans.ID, events.EventNameRequesterMsgReceived, eventData, w); err != nil {
 		http.Error(w, PublicFailedToProcessReqMsg, http.StatusInternalServerError)
-		return
 	}
-	var wg sync.WaitGroup
-	wg.Add(1)
-	waitingReqs[eventId] = RequestWait{
-		w:  &w,
-		wg: &wg,
-	}
-	wg.Wait()
 }
 
 func getRequesterMessageSupplier(ctx common.ExtendedContext, repo ill_db.IllRepo, illTransID string, symbol string) (*ill_db.LocatedSupplier, error) {
@@ -795,18 +786,9 @@ func handleSupplyingAgencyMessage(ctx common.ExtendedContext, illMessage *iso186
 		return
 	}
 
-	eventId, err := createNotice(ctx, eventBus, illTrans.ID, events.EventNameSupplierMsgReceived, eventData, events.EventStatusSuccess)
-	if err != nil {
+	if err := waitForMessageConfirmation(ctx, eventBus, illTrans.ID, events.EventNameSupplierMsgReceived, eventData, w); err != nil {
 		http.Error(w, PublicFailedToProcessReqMsg, http.StatusInternalServerError)
-		return
-	}
-	var wg sync.WaitGroup
-	wg.Add(1)
-	waitingReqs[eventId] = RequestWait{
-		w:  &w,
-		wg: &wg,
 	}
-	wg.Wait()
 }
 
 func validateStatusAndReasonForMessage(ctx common.ExtendedContext, illMessage *iso18626.ISO18626Message, w http.ResponseWriter, eventData events.EventData, eventBus events.EventBus, illTrans ill_db.IllTransaction) (iso18626.TypeStatus, iso18626.TypeReasonForMessage, error) {
@@ -976,7 +958,7 @@ func (h *Iso18626Handler) ConfirmRequesterMsg(ctx common.ExtendedContext, event
 		})
 		return
 	}
-	if _, ok := waitingReqs[reqRequestEvent.ID]; !ok {
+	if !waitingReqs.contains(reqRequestEvent.ID) {
 		return // instance doesn't have the paused request
 	}
 	// instance has the event, process it
@@ -1013,7 +995,7 @@ func (h *Iso18626Handler) ConfirmSupplierMsg(ctx common.ExtendedContext, event e
 		})
 		return
 	}
-	if _, ok := waitingReqs[supRequestEvent.ID]; !ok {
+	if !waitingReqs.contains(supRequestEvent.ID) {
 		return // instance doesn't have the paused request
 	}
 	// instance has the event, process it
@@ -1080,11 +1062,10 @@ func (h *Iso18626Handler) handleConfirmRequesterMsgTask(ctx common.ExtendedConte
 
 func (c *Iso18626Handler) confirmSupplierResponse(ctx common.ExtendedContext, illTransId string, waitRequestId string, requesterIllMsg *iso18626.ISO18626Message,
 	supplierResult events.EventResult) (*iso18626.ISO18626Message, error) {
-	wait, ok := waitingReqs[waitRequestId]
+	wait, ok := waitingReqs.take(waitRequestId)
 	if !ok {
 		return nil, fmt.Errorf("waiting request '%s' not found", waitRequestId)
 	}
-	delete(waitingReqs, waitRequestId)
 	var errorMessage = ""
 	var errorType *iso18626.TypeErrorType
 	var messageStatus = iso18626.TypeMessageStatusOK
@@ -1144,11 +1125,10 @@ func (c *Iso18626Handler) confirmSupplierResponse(ctx common.ExtendedContext, il
 
 func (c *Iso18626Handler) confirmRequesterResponse(ctx common.ExtendedContext, illTransId string, waitRequestId string, supplierIllMsg *iso18626.ISO18626Message,
 	requesterResult events.EventResult) (*iso18626.ISO18626Message, error) {
-	wait, ok := waitingReqs[waitRequestId]
+	wait, ok := waitingReqs.take(waitRequestId)
 	if !ok {
 		return nil, fmt.Errorf("waiting request '%s' not found", waitRequestId)
 	}
-	delete(waitingReqs, waitRequestId)
 	var errorMessage = ""
 	var errorType *iso18626.TypeErrorType
 	var messageStatus iso18626.TypeMessageStatus
@@ -1207,3 +1187,46 @@ type RequestWait struct {
 	w  *http.ResponseWriter
 	wg *sync.WaitGroup
 }
+
+// requestWaitRegistry coordinates HTTP handlers and asynchronous confirmations.
+// Taking a waiter removes it atomically so only one confirmation can write to it.
+type requestWaitRegistry struct {
+	mu       sync.Mutex
+	requests map[string]RequestWait
+}
+
+func (r *requestWaitRegistry) register(id string, wait RequestWait) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.requests[id] = wait
+}
+
+func (r *requestWaitRegistry) contains(id string) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	_, ok := r.requests[id]
+	return ok
+}
+
+func (r *requestWaitRegistry) take(id string) (RequestWait, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	wait, ok := r.requests[id]
+	delete(r.requests, id)
+	return wait, ok
+}
+
+func waitForMessageConfirmation(ctx common.ExtendedContext, eventBus events.EventBus, illTransID string, eventName events.EventName, data events.EventData, w http.ResponseWriter) error {
+	id := uuid.NewString()
+	var wg sync.WaitGroup
+	wg.Add(1)
+	// Register before publishing: a consumer may confirm before creation returns.
+	waitingReqs.register(id, RequestWait{w: &w, wg: &wg})
+	defer waitingReqs.take(id)
+	if _, err := eventBus.CreateNoticeWithID(id, illTransID, eventName, data, events.EventStatusSuccess, events.EventDomainIllTransaction, events.SignalConsumers); err != nil {
+		ctx.Logger().Error(InternalFailedToCreateNotice, "error", err, "transactionId", illTransID)
+		return err
+	}
+	wg.Wait()
+	return nil
+}

broker/handler/request_wait_test.go

@@ -0,0 +1,133 @@
+package handler
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http/httptest"
+	"sync"
+	"testing"
+
+	"github.com/indexdata/crosslink/broker/common"
+	"github.com/indexdata/crosslink/broker/events"
+	"github.com/indexdata/crosslink/iso18626"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRequestWaitRegistryConcurrentAccess(t *testing.T) {
+	registry := requestWaitRegistry{requests: make(map[string]RequestWait)}
+	start := make(chan struct{})
+	var workers sync.WaitGroup
+	for i := range 64 {
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+			<-start
+			id := fmt.Sprint(i)
+			var pending sync.WaitGroup
+			registry.register(id, RequestWait{wg: &pending})
+			assert.True(t, registry.contains(id))
+			var takers sync.WaitGroup
+			results := make(chan bool, 2)
+			for range 2 {
+				takers.Add(1)
+				go func() {
+					defer takers.Done()
+					wait, ok := registry.take(id)
+					if ok {
+						assert.Same(t, &pending, wait.wg)
+					}
+					results <- ok
+				}()
+			}
+			takers.Wait()
+			assert.NotEqual(t, <-results, <-results, "exactly one caller must consume the waiter")
+			assert.False(t, registry.contains(id))
+		}()
+	}
+	close(start)
+	workers.Wait()
+	assert.Empty(t, registry.requests)
+}
+
+type immediateNoticeBus struct {
+	events.EventBus
+	publish func(string) error
+}
+
+func (b immediateNoticeBus) CreateNoticeWithID(eventID, _ string, _ events.EventName, _ events.EventData, _ events.EventStatus, _ events.EventDomain, _ events.SignalTarget) (string, error) {
+	return eventID, b.publish(eventID)
+}
+
+func TestWaitForMessageConfirmationBeforePublishReturns(t *testing.T) {
+	ctx := common.CreateExtCtxWithArgs(context.Background(), nil)
+	h := &Iso18626Handler{}
+	for _, name := range []events.EventName{events.EventNameRequesterMsgReceived, events.EventNameSupplierMsgReceived} {
+		t.Run(string(name), func(t *testing.T) {
+			// All publications overlap, and each confirmation runs before publication returns.
+			const requests = 32
+			ready := make(chan struct{}, requests)
+			release := make(chan struct{})
+			var workers sync.WaitGroup
+			for range requests {
+				workers.Add(1)
+				go func() {
+					defer workers.Done()
+					recorder := httptest.NewRecorder()
+					bus := immediateNoticeBus{publish: func(id string) error {
+						assert.True(t, waitingReqs.contains(id))
+						ready <- struct{}{}
+						<-release
+						msg := iso18626.NewISO18626Message()
+						result := events.EventResult{CustomData: map[string]any{common.DO_NOT_SEND: true}}
+						var confirmation *iso18626.ISO18626Message
+						var err error
+						if name == events.EventNameRequesterMsgReceived {
+							msg.RequestingAgencyMessage = &iso18626.RequestingAgencyMessage{}
+							confirmation, err = h.confirmSupplierResponse(ctx, "transaction", id, msg, result)
+						} else {
+							msg.SupplyingAgencyMessage = &iso18626.SupplyingAgencyMessage{}
+							confirmation, err = h.confirmRequesterResponse(ctx, "transaction", id, msg, result)
+						}
+						if !assert.NoError(t, err) {
+							return err
+						}
+						assert.NotNil(t, confirmation)
+						assert.False(t, waitingReqs.contains(id))
+						// A duplicate confirmation must not write or complete the waiter twice.
+						if name == events.EventNameRequesterMsgReceived {
+							_, err = h.confirmSupplierResponse(ctx, "transaction", id, msg, result)
+						} else {
+							_, err = h.confirmRequesterResponse(ctx, "transaction", id, msg, result)
+						}
+						assert.ErrorContains(t, err, "not found")
+						return nil
+					}}
+					assert.NoError(t, waitForMessageConfirmation(ctx, bus, "transaction", name, events.EventData{}, recorder))
+					assert.Equal(t, 200, recorder.Code)
+					assert.Contains(t, recorder.Body.String(), "MessageConfirmation")
+				}()
+			}
+			for range requests {
+				<-ready
+			}
+			close(release)
+			workers.Wait()
+		})
+	}
+}
+
+func TestWaitForMessageConfirmationPublishFailure(t *testing.T) {
+	ctx := common.CreateExtCtxWithArgs(context.Background(), nil)
+	publishErr := errors.New("publication failed")
+	var eventID string
+	bus := immediateNoticeBus{publish: func(id string) error {
+		eventID = id
+		assert.True(t, waitingReqs.contains(id))
+		return publishErr
+	}}
+	err := waitForMessageConfirmation(ctx, bus, "transaction", events.EventNameRequesterMsgReceived, events.EventData{}, httptest.NewRecorder())
+	require.ErrorIs(t, err, publishErr)
+	assert.False(t, waitingReqs.contains(eventID), "failed publication must not leave a waiter")
+}

broker/patron_request/service/action_test.go

@@ -6280,3 +6280,8 @@ func (r *MockPrRepo) GetLendingPredecessorForUpdate(ctx common.ExtendedContext,
 	args := r.Called(params)
 	return args.Get(0).(pr_db.PatronRequest), args.Error(1)
 }
+
+func (m *MockEventBus) CreateNoticeWithID(eventID string, id string, eventName events.EventName, data events.EventData, status events.EventStatus, eventDomain events.EventDomain, target events.SignalTarget) (string, error) {
+	_, err := m.CreateNotice(id, eventName, data, status, eventDomain, target)
+	return eventID, err
+}