From 5b58baece53ac93af3acd72af9d968caa28dce1a Mon Sep 17 00:00:00 2001 From: Adam Dickmeiss Date: Thu, 8 Oct 2026 17:00:59 +0200 Subject: [PATCH] Fix races in ISO message confirmations ISO HTTP handlers and asynchronous confirmation handlers accessed the shared waitingReqs map without synchronization. Notices were also published before their waiters were registered, allowing a fast confirmation to arrive before the HTTP request was ready to receive it. Protect all map access with a mutex, consume waiters through atomic lookup and removal, and register waiters before publishing notices. Remove waiters when publication fails. Add regression tests covering concurrent access, early confirmations, duplicate confirmations, and publication failure. Verified that the tests detect both original defects. Validation: focused race tests passed for 20 repetitions; ISO integration race tests, the full broker suite, lint, build, and all 135 Bruno requests passed. --- broker/events/eventbus.go | 14 +- broker/events/eventbus_test.go | 37 ++++++ broker/handler/iso18626-handler.go | 77 +++++++---- broker/handler/request_wait_test.go | 133 +++++++++++++++++++ broker/patron_request/service/action_test.go | 5 + 5 files changed, 235 insertions(+), 31 deletions(-) create mode 100644 broker/handler/request_wait_test.go diff --git a/broker/events/eventbus.go b/broker/events/eventbus.go index 8a2be7979..9f031e2ef 100644 --- a/broker/events/eventbus.go +++ b/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{ diff --git a/broker/events/eventbus_test.go b/broker/events/eventbus_test.go index ba05cd251..35ccda05c 100644 --- a/broker/events/eventbus_test.go +++ b/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 := ¬iceIDRepo{} + 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 +} diff --git a/broker/handler/iso18626-handler.go b/broker/handler/iso18626-handler.go index 87c0a23fb..b99528fe4 100644 --- a/broker/handler/iso18626-handler.go +++ b/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 +} diff --git a/broker/handler/request_wait_test.go b/broker/handler/request_wait_test.go new file mode 100644 index 000000000..eaa153dbf --- /dev/null +++ b/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") +} diff --git a/broker/patron_request/service/action_test.go b/broker/patron_request/service/action_test.go index 80b65d3f2..0fb7ac209 100644 --- a/broker/patron_request/service/action_test.go +++ b/broker/patron_request/service/action_test.go @@ -6252,3 +6252,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 +}