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 2f7f01ef4..64c05d9c8 100644 --- a/broker/patron_request/service/action_test.go +++ b/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 +}