mirror/hatchet
1
0
Fork 0
mirror of https://github.com/hatchet-dev/hatchet.git synced 2026-05-06 18:09:49 -05:00
Code Issues Packages Projects Releases Wiki Activity

Fix: Add back task event tmp dual write (#3828)

Browse Source 
* fix: add back tmp event write

* fix: handle error
This commit is contained in:
matt
2026-05-05 16:02:34 -04:00
committed by GitHub
parent 5851bb029f
commit 0911c11fca
1 changed files with 18 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/repository/olap.go
+18
View File
@@ -1714,6 +1714,7 @@ func (r *OLAPRepositoryImpl) acquireAdvisoryLocksForWorkflowRuns(ctx context.Con
func (r *OLAPRepositoryImpl) writeTaskEventBatch(ctx context.Context, tenantId uuid.UUID, events []sqlcv1.CreateTaskEventsOLAPParams, workflowRunIds []uuid.UUID) (*StatusUpdateResult, error) {
eventsToWrite := make([]sqlcv1.CreateTaskEventsOLAPParams, 0)
eventsForStatusUpdate := make([]sqlcv1.CreateTaskEventsOLAPParams, 0, len(events))
tmpEventsToWrite := make([]sqlcv1.CreateTaskEventsOLAPTmpParams, 0)
payloadsToWrite := make([]StoreOLAPPayloadOpts, 0)
for _, event := range events {
@@ -1726,6 +1727,16 @@ func (r *OLAPRepositoryImpl) writeTaskEventBatch(ctx context.Context, tenantId u
}
eventsToWrite = append(eventsToWrite, event)
tmpEventsToWrite = append(tmpEventsToWrite, sqlcv1.CreateTaskEventsOLAPTmpParams{
TenantID: event.TenantID,
TaskID: event.TaskID,
TaskInsertedAt: event.TaskInsertedAt,
EventType: event.EventType,
RetryCount: event.RetryCount,
ReadableStatus: event.ReadableStatus,
WorkerID: event.WorkerID,
})
}
eventsForStatusUpdate = append(eventsForStatusUpdate, event)
@@ -1762,6 +1773,13 @@ func (r *OLAPRepositoryImpl) writeTaskEventBatch(ctx context.Context, tenantId u
}
}
if len(tmpEventsToWrite) > 0 {
_, err = r.queries.CreateTaskEventsOLAPTmp(ctx, tx, tmpEventsToWrite)
if err != nil {
return nil, err
}
}
statusUpdates := r.prepareStatusUpdateBatch(ctx, tenantId, eventsForStatusUpdate)
result := &StatusUpdateResult{}