mirror/hatchet
1
0
Fork 0
mirror of https://github.com/hatchet-dev/hatchet.git synced 2025-12-23 01:30:17 -06:00
Code Issues Packages Projects Releases Wiki Activity

Fix: Streaming Bugs (#1913)

Browse Source 
* fix: bug with json parsing failing

* fix: hang up on cancel and fail

* fix: pub stream events even if tenant pubs are disabled

* fix: condition

* fix: eq
This commit is contained in:
Matt Kaye
2025-06-26 16:22:56 -04:00
committed by GitHub
parent b92c1588ba
commit e62f7edab3
3 changed files with 7 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
internal/msgqueue/v1/rabbitmq/rabbitmq.go
View File

@@ -231,7 +231,7 @@ func (t *MessageQueueImpl) pubMessage(ctx context.Context, q msgqueue.Queue, msg
}
// if this is a tenant msg, publish to the tenant exchange
if !t.disableTenantExchangePubs && msg.TenantID != "" {
if (!t.disableTenantExchangePubs || msg.ID == "task-stream-event") && msg.TenantID != "" {
// determine if the tenant exchange exists
if _, ok := t.tenantIdCache.Get(msg.TenantID); !ok {
// register the tenant exchange

2
internal/services/dispatcher/server_v1.go
View File

@@ -923,7 +923,7 @@ func (s *DispatcherImpl) subscribeToWorkflowEventsByWorkflowRunIdV1(workflowRunI
}
isWorkflowRunCompletedEvent := e.ResourceType == contracts.ResourceType_RESOURCE_TYPE_WORKFLOW_RUN &&
e.EventType == contracts.ResourceEventType_RESOURCE_EVENT_TYPE_COMPLETED
(e.EventType == contracts.ResourceEventType_RESOURCE_EVENT_TYPE_COMPLETED || e.EventType == contracts.ResourceEventType_RESOURCE_EVENT_TYPE_FAILED || e.EventType == contracts.ResourceEventType_RESOURCE_EVENT_TYPE_CANCELLED)
if isWorkflowRunCompletedEvent {
e.Hangup = true

5
pkg/repository/v1/olap.go
View File

@@ -417,6 +417,11 @@ type TaskMetadata struct {
func ParseTaskMetadata(jsonData []byte) ([]TaskMetadata, error) {
var tasks []TaskMetadata
if len(jsonData) == 0 {
return tasks, nil
}
err := json.Unmarshal(jsonData, &tasks)
if err != nil {
return nil, err