Skip to content

ILLDEV-463 Add batch action cleanup #692

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JanisSaldabols merged 4 commits into from
Jul 30, 2026
Merged

ILLDEV-463 Add batch action cleanup #692

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
559f18e
ILLDEV-463 Add batch action cleanup
JanisSaldabols Jul 27, 2026
efdf715
ILLDEV-456 Filter only finished events and not current event
JanisSaldabols Jul 28, 2026
3b7c419
ILLDEV-463 Delete only events which don't have patron request
JanisSaldabols Jul 28, 2026
b6db822
ILLDEV-463 Clean parent column
JanisSaldabols Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 52 additions & 51 deletions broker/README.md
View file
Open in desktop

Large diffs are not rendered by default.

8 changes: 4 additions & 4 deletions broker/app/app.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,7 @@ func Init(ctx context.Context) (Context, error) {
illRepo := ill_db.CreateIllRepo(pool)
prRepo := pr_db.CreatePrRepo(pool, DB_EXPLAIN_ANALYZE)
psRepo := ps_db.CreatePsRepo(pool)
schedRepo := sched_db.CreateSchedRepo(pool)

var emailSenderService *sched_service.EmailSenderService
emailSenderService, err = sched_service.NewEmailSenderService(prRepo, illRepo)
Expand All @@ -206,7 +207,7 @@ func Init(ctx context.Context) (Context, error) {
sseBroker := api.NewSseBroker(appCtx, tenantResolver)
psApiHandler := psapi.NewPsApiHandler(psRepo, prRepo, tenantResolver)

batchActionService := sched_service.NewBatchActionService(eventBus, prRepo, emailSenderService)
batchActionService := sched_service.NewBatchActionService(eventBus, prRepo, schedRepo, emailSenderService)

if err != nil {
appCtx.Logger().Warn("email service not available, email sending events will fail", "error", err)
Expand All @@ -218,9 +219,8 @@ func Init(ctx context.Context) (Context, error) {
return Context{}, err
}

schedRepoRepo := sched_db.CreateSchedRepo(pool)
schedApiHandler := schedapi.NewSchedulerApiHandler(API_PAGE_SIZE, schedRepoRepo, eventRepo, tenantResolver)
if err = StartScheduler(ctx, schedRepoRepo, eventBus); err != nil {
schedApiHandler := schedapi.NewSchedulerApiHandler(API_PAGE_SIZE, schedRepo, eventRepo, tenantResolver)
if err = StartScheduler(ctx, schedRepo, eventBus); err != nil {
return Context{}, err
}

Expand Down
11 changes: 11 additions & 0 deletions broker/common/common.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package common
import (
"encoding/json"
"fmt"
"math"
"reflect"
"strings"
)
Expand Down Expand Up @@ -133,3 +134,13 @@ func SplitAgencySymbol(symbol string) (string, string) {
}
return symbolParts[0], symbolParts[1]
}

func ToInt32(i int) int32 {
if i > math.MaxInt32 {
return math.MaxInt32
} else if i < math.MinInt32 {
return math.MinInt32
} else {
return int32(i)
}
}
10 changes: 10 additions & 0 deletions broker/scheduler/db/repo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ type SchedRepo interface {
GetScheduledTaskByIdForUpdate(ctx common.ExtendedContext, id string, owners []string) (ScheduledTask, error)
HasActiveBatchActionEvents(ctx common.ExtendedContext, taskID string) (bool, error)
DeleteBatchActionEvents(ctx common.ExtendedContext, taskID string) error
DeleteOldBatchActionRunEvents(ctx common.ExtendedContext, currentEventId string, taskID string, retention int32) error
DeleteScheduledTask(ctx common.ExtendedContext, id string, owners []string) error
GetScheduledTasks(ctx common.ExtendedContext, params GetScheduledTasksParams) ([]ScheduledTask, int64, error)
}
Expand Down Expand Up @@ -127,6 +128,15 @@ func (r *PgSchedRepo) DeleteBatchActionEvents(ctx common.ExtendedContext, taskID
return r.eventQueries.DeleteBatchActionEvents(ctx, r.GetConnOrTx(), taskID)
}

func (r *PgSchedRepo) DeleteOldBatchActionRunEvents(ctx common.ExtendedContext, currentEventId string, taskID string, retention int32) error {
return r.eventQueries.DeleteOldBatchActionRunEvents(ctx, r.GetConnOrTx(), events.DeleteOldBatchActionRunEventsParams{
CurrentEventID: currentEventId,
TaskID: taskID,
Retention: retention,
PatronRequestID: events.DEFAULT_PATRON_REQUEST_ID,
Comment thread
JanisSaldabols marked this conversation as resolved.
})
}

func (r *PgSchedRepo) GetScheduledTasks(ctx common.ExtendedContext, params GetScheduledTasksParams) ([]ScheduledTask, int64, error) {
rows, err := r.queries.GetScheduledTasks(ctx, r.GetConnOrTx(), params)
var tasks []ScheduledTask
Expand Down
22 changes: 21 additions & 1 deletion broker/scheduler/service/batch_action.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,23 +11,29 @@ import (
"github.com/indexdata/crosslink/broker/events"
pr_db "github.com/indexdata/crosslink/broker/patron_request/db"
prservice "github.com/indexdata/crosslink/broker/patron_request/service"
sched_db "github.com/indexdata/crosslink/broker/scheduler/db"
schedoapi "github.com/indexdata/crosslink/broker/scheduler/oapi"
"github.com/indexdata/go-utils/utils"
)

const BATCH_COMP = "batch_action"
const TIME_FORMAT = "2006-01-02 15:04:05"

var BATCH_ACTION_RUN_RETENTION = common.ToInt32(utils.Must(utils.GetEnvInt("BATCH_ACTION_RUN_RETENTION", 5)))

type BatchActionService struct {
eventBus events.EventBus
prRepo pr_db.PrRepo
schedRepo sched_db.SchedRepo
emailSenderService *EmailSenderService
actionMappingService prservice.ActionMappingService
}

func NewBatchActionService(eventBus events.EventBus, prRepo pr_db.PrRepo, emailSenderService *EmailSenderService) *BatchActionService {
func NewBatchActionService(eventBus events.EventBus, prRepo pr_db.PrRepo, schedRepo sched_db.SchedRepo, emailSenderService *EmailSenderService) *BatchActionService {
return &BatchActionService{
eventBus: eventBus,
prRepo: prRepo,
schedRepo: schedRepo,
emailSenderService: emailSenderService,
actionMappingService: prservice.ActionMappingService{SMService: &prservice.StateModelService{}},
}
Expand Down Expand Up @@ -70,9 +76,23 @@ func (s *BatchActionService) batchAction(ctx common.ExtendedContext, event event
batchActionData := *event.EventData.BatchActionData
batchActionData.Selector = restrictedSelector
event.EventData.BatchActionData = &batchActionData
s.cleanupOldRuns(ctx, event)
return action(ctx, event)
}

func (s *BatchActionService) cleanupOldRuns(ctx common.ExtendedContext, event events.Event) {
if BATCH_ACTION_RUN_RETENTION <= 0 {
return
}
taskID := event.EventData.BatchActionData.TaskId
if taskID == "" {
return
}
if err := s.schedRepo.DeleteOldBatchActionRunEvents(ctx, event.ID, taskID, BATCH_ACTION_RUN_RETENTION); err != nil {
ctx.Logger().Error("failed to cleanup old batch action runs", "taskId", taskID, "retention", BATCH_ACTION_RUN_RETENTION, "error", err)
}
}

func addBatchActionOwnerRestriction(selector string, owner string) (string, error) {
if selector == "" {
return "", fmt.Errorf("selector is empty")
Expand Down
94 changes: 80 additions & 14 deletions broker/scheduler/service/batch_action_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (
"github.com/indexdata/crosslink/broker/events"
pr_db "github.com/indexdata/crosslink/broker/patron_request/db"
prservice "github.com/indexdata/crosslink/broker/patron_request/service"
sched_db "github.com/indexdata/crosslink/broker/scheduler/db"
schedoapi "github.com/indexdata/crosslink/broker/scheduler/oapi"
"github.com/stretchr/testify/assert"
)
Expand Down Expand Up @@ -39,6 +40,23 @@ type mockBatchActionEventBus struct {
createTaskErrByID map[string]error
}

type mockBatchActionCleanupRepo struct {
sched_db.PgSchedRepo
called bool
gotCurrentEventId string
gotTaskID string
gotRetention int32
err error
}

func (m *mockBatchActionCleanupRepo) DeleteOldBatchActionRunEvents(_ common.ExtendedContext, currentEventId string, taskID string, retention int32) error {
m.called = true
m.gotCurrentEventId = currentEventId
m.gotTaskID = taskID
m.gotRetention = retention
return m.err
}

func (m *mockBatchActionEventBus) ProcessTask(
ctx common.ExtendedContext,
event events.Event,
Expand Down Expand Up @@ -99,7 +117,7 @@ func TestNewBatchActionService_WiresDependencies(t *testing.T) {
eventBus := &mockBatchActionEventBus{}
emailSender := EmailSenderServiceWithClient(nil, nil, nil, nil)

svc := NewBatchActionService(eventBus, &mockEmailPrRepo{}, emailSender)
svc := NewBatchActionService(eventBus, &mockEmailPrRepo{}, &mockBatchActionCleanupRepo{}, emailSender)

assert.NotNil(t, svc)
assert.Same(t, eventBus, svc.eventBus)
Expand All @@ -109,7 +127,7 @@ func TestNewBatchActionService_WiresDependencies(t *testing.T) {
func TestBatchAction_CallsProcessTaskWithSignalConsumers(t *testing.T) {
eventBus := &mockBatchActionEventBus{}
emailSender := EmailSenderServiceWithClient(nil, nil, nil, nil)
svc := NewBatchActionService(eventBus, &mockEmailPrRepo{}, emailSender)
svc := NewBatchActionService(eventBus, &mockEmailPrRepo{}, &mockBatchActionCleanupRepo{}, emailSender)

event := events.Event{}

Expand All @@ -129,7 +147,7 @@ func TestBatchAction_ProcessTaskErrorIgnored(t *testing.T) {
processErr: errors.New("event bus unavailable"),
}
emailSender := EmailSenderServiceWithClient(nil, nil, nil, nil)
svc := NewBatchActionService(eventBus, &mockEmailPrRepo{}, emailSender)
svc := NewBatchActionService(eventBus, &mockEmailPrRepo{}, &mockBatchActionCleanupRepo{}, emailSender)

assert.NotPanics(t, func() {
svc.BatchAction(testCtx, events.Event{})
Expand All @@ -138,8 +156,56 @@ func TestBatchAction_ProcessTaskErrorIgnored(t *testing.T) {
assert.True(t, eventBus.processCalled)
}

func TestBatchAction_CleansOldRunsBeforeDispatch(t *testing.T) {
repo := &mockEmailPrRepo{}
eventBus := &mockBatchActionEventBus{}
cleanupRepo := &mockBatchActionCleanupRepo{}
svc := NewBatchActionService(eventBus, repo, cleanupRepo, nil)

status, result := svc.batchAction(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"}))

assert.Equal(t, events.EventStatusSuccess, status)
assert.NotNil(t, result)
assert.True(t, cleanupRepo.called)
assert.Equal(t, "batch-event-1", cleanupRepo.gotCurrentEventId)
assert.Equal(t, "task-1", cleanupRepo.gotTaskID)
assert.Equal(t, int32(5), cleanupRepo.gotRetention)
assert.True(t, repo.listCalled)
}
Comment thread
JanisSaldabols marked this conversation as resolved.

func TestBatchAction_CleansOldRunsBeforeDispatch_WhenRetentionIsZero(t *testing.T) {
repo := &mockEmailPrRepo{}
eventBus := &mockBatchActionEventBus{}
cleanupRepo := &mockBatchActionCleanupRepo{}
svc := NewBatchActionService(eventBus, repo, cleanupRepo, nil)
prevValue := BATCH_ACTION_RUN_RETENTION
BATCH_ACTION_RUN_RETENTION = 0

status, result := svc.batchAction(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"}))

assert.Equal(t, events.EventStatusSuccess, status)
assert.NotNil(t, result)
assert.False(t, cleanupRepo.called)
assert.True(t, repo.listCalled)
BATCH_ACTION_RUN_RETENTION = prevValue
}

func TestBatchAction_CleanupErrorDoesNotBlockDispatch(t *testing.T) {
repo := &mockEmailPrRepo{}
eventBus := &mockBatchActionEventBus{}
cleanupRepo := &mockBatchActionCleanupRepo{err: errors.New("delete failed")}
svc := NewBatchActionService(eventBus, repo, cleanupRepo, nil)

status, result := svc.batchAction(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"}))

assert.Equal(t, events.EventStatusSuccess, status)
assert.NotNil(t, result)
assert.True(t, cleanupRepo.called)
assert.True(t, repo.listCalled)
}

func TestBatchAction_NilBatchActionDataReturnsError(t *testing.T) {
svc := NewBatchActionService(nil, &mockEmailPrRepo{}, nil)
svc := NewBatchActionService(nil, &mockEmailPrRepo{}, &mockBatchActionCleanupRepo{}, nil)

status, result := svc.batchAction(testCtx, events.Event{})

Expand All @@ -151,7 +217,7 @@ func TestBatchAction_NilBatchActionDataReturnsError(t *testing.T) {
}

func TestBatchAction_UnknownActionReturnsError(t *testing.T) {
svc := NewBatchActionService(nil, &mockEmailPrRepo{}, nil)
svc := NewBatchActionService(nil, &mockEmailPrRepo{}, &mockBatchActionCleanupRepo{}, nil)

event := batchActionEvent("unknown-action")

Expand All @@ -166,7 +232,7 @@ func TestBatchAction_UnknownActionReturnsError(t *testing.T) {

func TestBatchAction_EmailPullslipsDispatchesToEmailSender(t *testing.T) {
emailSender := EmailSenderServiceWithClient(nil, nil, &mockEmailService{ready: false}, nil)
svc := NewBatchActionService(nil, &mockEmailPrRepo{}, emailSender)
svc := NewBatchActionService(nil, &mockEmailPrRepo{}, &mockBatchActionCleanupRepo{}, emailSender)

event := batchActionEvent(string(schedoapi.EmailPullslips))

Expand All @@ -181,7 +247,7 @@ func TestBatchAction_EmailPullslipsDispatchesToEmailSender(t *testing.T) {

func TestBatchAction_RequestAgingDispatches(t *testing.T) {
repo := &mockEmailPrRepo{}
svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, nil)
svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, &mockBatchActionCleanupRepo{}, nil)

status, result := svc.batchAction(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"}))

Expand All @@ -193,7 +259,7 @@ func TestBatchAction_RequestAgingDispatches(t *testing.T) {

func TestBatchAction_AddsOwnerRestrictionBeforeDispatch(t *testing.T) {
repo := &mockEmailPrRepo{}
svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, nil)
svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, &mockBatchActionCleanupRepo{}, nil)
event := requestAgingEvent("state = REQ", map[string]any{"interval": "24h"})

status, _ := svc.batchAction(testCtx, event)
Expand All @@ -210,7 +276,7 @@ func TestBatchAction_AddsOwnerRestrictionBeforeDispatch(t *testing.T) {

func TestBatchAction_MissingOwnerDispatchesWithoutRestriction(t *testing.T) {
repo := &mockEmailPrRepo{}
svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, nil)
svc := NewBatchActionService(&mockBatchActionEventBus{}, repo, &mockBatchActionCleanupRepo{}, nil)
event := requestAgingEvent("state = REQ", map[string]any{"interval": "24h"})
event.EventData.BatchActionData.Owner = ""

Expand Down Expand Up @@ -290,7 +356,7 @@ func TestRequestAging_ValidationErrors(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
repo := &mockEmailPrRepo{}
eventBus := &mockBatchActionEventBus{}
svc := NewBatchActionService(eventBus, repo, nil)
svc := NewBatchActionService(eventBus, repo, &mockBatchActionCleanupRepo{}, nil)

status, result := svc.RequestAging(testCtx, tt.event)

Expand All @@ -312,7 +378,7 @@ func TestRequestAging_ValidationErrors(t *testing.T) {
func TestRequestAging_NoPatronRequestsReturnsSuccess(t *testing.T) {
repo := &mockEmailPrRepo{listResult: []pr_db.PatronRequest{}}
eventBus := &mockBatchActionEventBus{}
svc := NewBatchActionService(eventBus, repo, nil)
svc := NewBatchActionService(eventBus, repo, &mockBatchActionCleanupRepo{}, nil)

status, result := svc.RequestAging(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"}))

Expand All @@ -329,7 +395,7 @@ func TestRequestAging_CreatesBackgroundTasksForLending(t *testing.T) {
{ID: "lending-2", Side: prservice.SideLending, State: prservice.LenderStateWillSupply},
}}
eventBus := &mockBatchActionEventBus{}
svc := NewBatchActionService(eventBus, repo, nil)
svc := NewBatchActionService(eventBus, repo, &mockBatchActionCleanupRepo{}, nil)

status, result := svc.RequestAging(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h", "note": "Closing stale request", "reasonUnfilled": "Expired"}))

Expand All @@ -354,7 +420,7 @@ func TestRequestAging_FailsIfNoClosingActionForState(t *testing.T) {
{ID: "lending-2", Side: prservice.SideLending, State: prservice.LenderStateWillSupply},
}}
eventBus := &mockBatchActionEventBus{}
svc := NewBatchActionService(eventBus, repo, nil)
svc := NewBatchActionService(eventBus, repo, &mockBatchActionCleanupRepo{}, nil)

status, result := svc.RequestAging(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h", "note": "Closing stale request", "reasonUnfilled": "Expired"}))

Expand All @@ -371,7 +437,7 @@ func TestRequestAging_CreateTaskErrorRecordsCustomDataAndContinues(t *testing.T)
{ID: "ok-1", Side: prservice.SideLending, State: prservice.LenderStateWillSupply},
}}
eventBus := &mockBatchActionEventBus{createTaskErrByID: map[string]error{"failed-1": errors.New("create failed")}}
svc := NewBatchActionService(eventBus, repo, nil)
svc := NewBatchActionService(eventBus, repo, &mockBatchActionCleanupRepo{}, nil)

status, result := svc.RequestAging(testCtx, requestAgingEvent("cql.allRecords=1", map[string]any{"interval": "24h"}))

Expand Down
12 changes: 1 addition & 11 deletions broker/service/supplierlocator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -277,7 +277,7 @@ func (s *SupplierLocator) locateSuppliers(ctx common.ExtendedContext, event even
if pass != matchPass {
continue
}
added, loopErr := s.addLocatedSupplier(ctx, illTrans.ID, ToInt32(i), &sup)
added, loopErr := s.addLocatedSupplier(ctx, illTrans.ID, common.ToInt32(i), &sup)
i++
if loopErr == nil {
locatedSuppliers = append(locatedSuppliers, added)
Expand Down Expand Up @@ -562,16 +562,6 @@ func getPeerRatio(peer ill_db.Peer) float32 {
}
}

func ToInt32(i int) int32 {
if i > math.MaxInt32 {
return math.MaxInt32
} else if i < math.MinInt32 {
return math.MinInt32
} else {
return int32(i)
}
}

type SkippedSupplier struct {
Symbol string `json:"symbol"`
Reason string `json:"reason"`
Expand Down
Loading
Loading