018dfb1b-c20b-4820-815a-b6aed230920a

Committed 01 Mar 2024 05:41PM UTC coverage: 62.893% (+0.03%) from 62.868%

Build # 018dfb1b-c20b-4820-815a-b6aed230920a

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Do not get workflow execution from database when shard is closed (#5697)

Run Details

7 of 7 new or added lines in 3 files covered. (100.0%)

66 existing lines in 10 files now uncovered.

92963 of 147811 relevant lines covered (62.89%)

2320.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

83.84

/common/persistence/sql/workflowStateMaps.go

// Copyright (c) 2017-2020 Uber Technologies, Inc.
// Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

package sql

import (
        "context"
        "database/sql"

        "github.com/uber/cadence/common"
        "github.com/uber/cadence/common/persistence"
        "github.com/uber/cadence/common/persistence/serialization"
        "github.com/uber/cadence/common/persistence/sql/sqlplugin"
        "github.com/uber/cadence/common/types"
)

func updateActivityInfos(
        ctx context.Context,
        tx sqlplugin.Tx,
        activityInfos []*persistence.InternalActivityInfo,
        deleteInfos []int64,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) error {

        if len(activityInfos) > 0 {
                rows := make([]sqlplugin.ActivityInfoMapsRow, len(activityInfos))
                for i, activityInfo := range activityInfos {
                        scheduledEvent, scheduledEncoding := persistence.FromDataBlob(activityInfo.ScheduledEvent)
                        startEvent, startEncoding := persistence.FromDataBlob(activityInfo.StartedEvent)

                        info := &serialization.ActivityInfo{
                                Version:                  activityInfo.Version,
                                ScheduledEventBatchID:    activityInfo.ScheduledEventBatchID,
                                ScheduledEvent:           scheduledEvent,
                                ScheduledEventEncoding:   scheduledEncoding,
                                ScheduledTimestamp:       activityInfo.ScheduledTime,
                                StartedID:                activityInfo.StartedID,
                                StartedEvent:             startEvent,
                                StartedEventEncoding:     startEncoding,
                                StartedTimestamp:         activityInfo.StartedTime,
                                ActivityID:               activityInfo.ActivityID,
                                RequestID:                activityInfo.RequestID,
                                ScheduleToStartTimeout:   activityInfo.ScheduleToStartTimeout,
                                ScheduleToCloseTimeout:   activityInfo.ScheduleToCloseTimeout,
                                StartToCloseTimeout:      activityInfo.StartToCloseTimeout,
                                HeartbeatTimeout:         activityInfo.HeartbeatTimeout,
                                CancelRequested:          activityInfo.CancelRequested,
                                CancelRequestID:          activityInfo.CancelRequestID,
                                TimerTaskStatus:          activityInfo.TimerTaskStatus,
                                Attempt:                  activityInfo.Attempt,
                                TaskList:                 activityInfo.TaskList,
                                StartedIdentity:          activityInfo.StartedIdentity,
                                HasRetryPolicy:           activityInfo.HasRetryPolicy,
                                RetryInitialInterval:     activityInfo.InitialInterval,
                                RetryBackoffCoefficient:  activityInfo.BackoffCoefficient,
                                RetryMaximumInterval:     activityInfo.MaximumInterval,
                                RetryExpirationTimestamp: activityInfo.ExpirationTime,
                                RetryMaximumAttempts:     activityInfo.MaximumAttempts,
                                RetryNonRetryableErrors:  activityInfo.NonRetriableErrors,
                                RetryLastFailureReason:   activityInfo.LastFailureReason,
                                RetryLastWorkerIdentity:  activityInfo.LastWorkerIdentity,
                                RetryLastFailureDetails:  activityInfo.LastFailureDetails,
                        }
                        blob, err := parser.ActivityInfoToBlob(info)
                        if err != nil {
                                return err
                        }
                        rows[i] = sqlplugin.ActivityInfoMapsRow{
                                ShardID:                  int64(shardID),
                                DomainID:                 domainID,
                                WorkflowID:               workflowID,
                                RunID:                    runID,
                                ScheduleID:               activityInfo.ScheduleID,
                                LastHeartbeatUpdatedTime: activityInfo.LastHeartBeatUpdatedTime,
                                LastHeartbeatDetails:     activityInfo.Details,
                                Data:                     blob.Data,
                                DataEncoding:             string(blob.Encoding),
                        }
                }

                if _, err := tx.ReplaceIntoActivityInfoMaps(ctx, rows); err != nil {
                        return convertCommonErrors(tx, "updateActivityInfos", "Failed to execute update query.", err)
                }
        }

        if len(deleteInfos) > 0 {
                if _, err := tx.DeleteFromActivityInfoMaps(ctx, &sqlplugin.ActivityInfoMapsFilter{
                        ShardID:     int64(shardID),
                        DomainID:    domainID,
                        WorkflowID:  workflowID,
                        RunID:       runID,
                        ScheduleIDs: deleteInfos,
                }); err != nil {
                        return convertCommonErrors(tx, "updateActivityInfos", "Failed to execute delete query.", err)
                }
        }

        return nil
}

func getActivityInfoMap(
        ctx context.Context,
        db sqlplugin.DB,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) (map[int64]*persistence.InternalActivityInfo, error) {

        rows, err := db.SelectFromActivityInfoMaps(ctx, &sqlplugin.ActivityInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        })
        if err != nil && err != sql.ErrNoRows {
                return nil, convertCommonErrors(db, "getActivityInfoMap", "", err)
        }

        ret := make(map[int64]*persistence.InternalActivityInfo)
        for _, row := range rows {
                decoded, err := parser.ActivityInfoFromBlob(row.Data, row.DataEncoding)
                if err != nil {
                        return nil, err
                }
                info := &persistence.InternalActivityInfo{
                        DomainID:                 row.DomainID.String(),
                        ScheduleID:               row.ScheduleID,
                        Details:                  row.LastHeartbeatDetails,
                        LastHeartBeatUpdatedTime: row.LastHeartbeatUpdatedTime,
                        Version:                  decoded.GetVersion(),
                        ScheduledEventBatchID:    decoded.GetScheduledEventBatchID(),
                        ScheduledEvent:           persistence.NewDataBlob(decoded.ScheduledEvent, common.EncodingType(decoded.GetScheduledEventEncoding())),
                        ScheduledTime:            decoded.GetScheduledTimestamp(),
                        StartedID:                decoded.GetStartedID(),
                        StartedTime:              decoded.GetStartedTimestamp(),
                        ActivityID:               decoded.GetActivityID(),
                        RequestID:                decoded.GetRequestID(),
                        ScheduleToStartTimeout:   decoded.GetScheduleToStartTimeout(),
                        ScheduleToCloseTimeout:   decoded.GetScheduleToCloseTimeout(),
                        StartToCloseTimeout:      decoded.GetStartToCloseTimeout(),
                        HeartbeatTimeout:         decoded.GetHeartbeatTimeout(),
                        CancelRequested:          decoded.GetCancelRequested(),
                        CancelRequestID:          decoded.GetCancelRequestID(),
                        TimerTaskStatus:          decoded.GetTimerTaskStatus(),
                        Attempt:                  decoded.GetAttempt(),
                        StartedIdentity:          decoded.GetStartedIdentity(),
                        TaskList:                 decoded.GetTaskList(),
                        HasRetryPolicy:           decoded.GetHasRetryPolicy(),
                        InitialInterval:          decoded.GetRetryInitialInterval(),
                        BackoffCoefficient:       decoded.GetRetryBackoffCoefficient(),
                        MaximumInterval:          decoded.GetRetryMaximumInterval(),
                        ExpirationTime:           decoded.GetRetryExpirationTimestamp(),
                        MaximumAttempts:          decoded.GetRetryMaximumAttempts(),
                        NonRetriableErrors:       decoded.GetRetryNonRetryableErrors(),
                        LastFailureReason:        decoded.GetRetryLastFailureReason(),
                        LastWorkerIdentity:       decoded.GetRetryLastWorkerIdentity(),
                        LastFailureDetails:       decoded.GetRetryLastFailureDetails(),
                }
                if decoded.StartedEvent != nil {
                        info.StartedEvent = persistence.NewDataBlob(decoded.StartedEvent, common.EncodingType(decoded.GetStartedEventEncoding()))
                }
                ret[row.ScheduleID] = info
        }

        return ret, nil
}

func deleteActivityInfoMap(
        ctx context.Context,
        tx sqlplugin.Tx,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
) error {

        if _, err := tx.DeleteFromActivityInfoMaps(ctx, &sqlplugin.ActivityInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        }); err != nil {
                return convertCommonErrors(tx, "deleteActivityInfoMap", "", err)
        }
        return nil
}

func updateTimerInfos(
        ctx context.Context,
        tx sqlplugin.Tx,
        timerInfos []*persistence.TimerInfo,
        deleteInfos []string,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) error {

        if len(timerInfos) > 0 {
                rows := make([]sqlplugin.TimerInfoMapsRow, len(timerInfos))
                for i, timerInfo := range timerInfos {
                        blob, err := parser.TimerInfoToBlob(&serialization.TimerInfo{
                                Version:         timerInfo.Version,
                                StartedID:       timerInfo.StartedID,
                                ExpiryTimestamp: timerInfo.ExpiryTime,
                                // TaskID is a misleading variable, it actually serves
                                // the purpose of indicating whether a timer task is
                                // generated for this timer info
                                TaskID: timerInfo.TaskStatus,
                        })
                        if err != nil {
                                return err
                        }
                        rows[i] = sqlplugin.TimerInfoMapsRow{
                                ShardID:      int64(shardID),
                                DomainID:     domainID,
                                WorkflowID:   workflowID,
                                RunID:        runID,
                                TimerID:      timerInfo.TimerID,
                                Data:         blob.Data,
                                DataEncoding: string(blob.Encoding),
                        }
                }
                if _, err := tx.ReplaceIntoTimerInfoMaps(ctx, rows); err != nil {
                        return convertCommonErrors(tx, "updateTimerInfos", "Failed to execute update query.", err)
                }
        }

        if len(deleteInfos) > 0 {
                if _, err := tx.DeleteFromTimerInfoMaps(ctx, &sqlplugin.TimerInfoMapsFilter{
                        ShardID:    int64(shardID),
                        DomainID:   domainID,
                        WorkflowID: workflowID,
                        RunID:      runID,
                        TimerIDs:   deleteInfos,
                }); err != nil {
                        return convertCommonErrors(tx, "updateTimerInfos", "Failed to execute delete query.", err)
                }
        }

        return nil
}

func getTimerInfoMap(
        ctx context.Context,
        db sqlplugin.DB,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) (map[string]*persistence.TimerInfo, error) {

        rows, err := db.SelectFromTimerInfoMaps(ctx, &sqlplugin.TimerInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        })
        if err != nil && err != sql.ErrNoRows {
                return nil, convertCommonErrors(db, "getTimerInfoMap", "", err)
        }
        ret := make(map[string]*persistence.TimerInfo)
        for _, row := range rows {
                info, err := parser.TimerInfoFromBlob(row.Data, row.DataEncoding)
                if err != nil {
                        return nil, err
                }
                ret[row.TimerID] = &persistence.TimerInfo{
                        TimerID:    row.TimerID,
                        Version:    info.GetVersion(),
                        StartedID:  info.GetStartedID(),
                        ExpiryTime: info.GetExpiryTimestamp(),
                        // TaskID is a misleading variable, it actually serves
                        // the purpose of indicating whether a timer task is
                        // generated for this timer info
                        TaskStatus: info.GetTaskID(),
                }
        }

        return ret, nil
}

func deleteTimerInfoMap(
        ctx context.Context,
        tx sqlplugin.Tx,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
) error {

        if _, err := tx.DeleteFromTimerInfoMaps(ctx, &sqlplugin.TimerInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        }); err != nil {
                return convertCommonErrors(tx, "deleteTimerInfoMap", "", err)
        }
        return nil
}

func updateChildExecutionInfos(
        ctx context.Context,
        tx sqlplugin.Tx,
        childExecutionInfos []*persistence.InternalChildExecutionInfo,
        deleteInfos []int64,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) error {

        if len(childExecutionInfos) > 0 {
                rows := make([]sqlplugin.ChildExecutionInfoMapsRow, len(childExecutionInfos))
                for i, childExecutionInfo := range childExecutionInfos {
                        initiateEvent, initiateEncoding := persistence.FromDataBlob(childExecutionInfo.InitiatedEvent)
                        startEvent, startEncoding := persistence.FromDataBlob(childExecutionInfo.StartedEvent)

                        info := &serialization.ChildExecutionInfo{
                                Version:                childExecutionInfo.Version,
                                InitiatedEventBatchID:  childExecutionInfo.InitiatedEventBatchID,
                                InitiatedEvent:         initiateEvent,
                                InitiatedEventEncoding: initiateEncoding,
                                StartedEvent:           startEvent,
                                StartedEventEncoding:   startEncoding,
                                StartedID:              childExecutionInfo.StartedID,
                                StartedWorkflowID:      childExecutionInfo.StartedWorkflowID,
                                StartedRunID:           serialization.MustParseUUID(childExecutionInfo.StartedRunID),
                                CreateRequestID:        childExecutionInfo.CreateRequestID,
                                DomainID:               childExecutionInfo.DomainID,
                                DomainNameDEPRECATED:   childExecutionInfo.DomainNameDEPRECATED,
                                WorkflowTypeName:       childExecutionInfo.WorkflowTypeName,
                                ParentClosePolicy:      int32(childExecutionInfo.ParentClosePolicy),
                        }
                        blob, err := parser.ChildExecutionInfoToBlob(info)
                        if err != nil {
                                return err
                        }
                        rows[i] = sqlplugin.ChildExecutionInfoMapsRow{
                                ShardID:      int64(shardID),
                                DomainID:     domainID,
                                WorkflowID:   workflowID,
                                RunID:        runID,
                                InitiatedID:  childExecutionInfo.InitiatedID,
                                Data:         blob.Data,
                                DataEncoding: string(blob.Encoding),
                        }
                }
                if _, err := tx.ReplaceIntoChildExecutionInfoMaps(ctx, rows); err != nil {
                        return convertCommonErrors(tx, "updateChildExecutionInfos", "Failed to execute update query.", err)
                }
        }

        if len(deleteInfos) > 0 {
                if _, err := tx.DeleteFromChildExecutionInfoMaps(ctx, &sqlplugin.ChildExecutionInfoMapsFilter{
                        ShardID:      int64(shardID),
                        DomainID:     domainID,
                        WorkflowID:   workflowID,
                        RunID:        runID,
                        InitiatedIDs: deleteInfos,
                }); err != nil {
                        return convertCommonErrors(tx, "updateChildExecutionInfos", "Failed to execute delete query.", err)
                }
        }

        return nil
}

func getChildExecutionInfoMap(
        ctx context.Context,
        db sqlplugin.DB,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) (map[int64]*persistence.InternalChildExecutionInfo, error) {

        rows, err := db.SelectFromChildExecutionInfoMaps(ctx, &sqlplugin.ChildExecutionInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        })
        if err != nil && err != sql.ErrNoRows {
                return nil, convertCommonErrors(db, "getChildExecutionInfoMap", "", err)
        }

        ret := make(map[int64]*persistence.InternalChildExecutionInfo)
        for _, row := range rows {
                rowInfo, err := parser.ChildExecutionInfoFromBlob(row.Data, row.DataEncoding)
                if err != nil {
                        return nil, err
                }
                info := &persistence.InternalChildExecutionInfo{
                        InitiatedID:           row.InitiatedID,
                        InitiatedEventBatchID: rowInfo.GetInitiatedEventBatchID(),
                        Version:               rowInfo.GetVersion(),
                        StartedID:             rowInfo.GetStartedID(),
                        StartedWorkflowID:     rowInfo.GetStartedWorkflowID(),
                        StartedRunID:          serialization.UUID(rowInfo.GetStartedRunID()).String(),
                        CreateRequestID:       rowInfo.GetCreateRequestID(),
                        DomainID:              rowInfo.GetDomainID(),
                        DomainNameDEPRECATED:  rowInfo.GetDomainNameDEPRECATED(),
                        WorkflowTypeName:      rowInfo.GetWorkflowTypeName(),
                        ParentClosePolicy:     types.ParentClosePolicy(rowInfo.GetParentClosePolicy()),
                }
                if rowInfo.InitiatedEvent != nil {
                        info.InitiatedEvent = persistence.NewDataBlob(rowInfo.InitiatedEvent, common.EncodingType(rowInfo.GetInitiatedEventEncoding()))
                }
                if rowInfo.StartedEvent != nil {
                        info.StartedEvent = persistence.NewDataBlob(rowInfo.StartedEvent, common.EncodingType(rowInfo.GetStartedEventEncoding()))
                }
                ret[row.InitiatedID] = info
        }

        return ret, nil
}

func deleteChildExecutionInfoMap(
        ctx context.Context,
        tx sqlplugin.Tx,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
) error {

        if _, err := tx.DeleteFromChildExecutionInfoMaps(ctx, &sqlplugin.ChildExecutionInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        }); err != nil {
                return convertCommonErrors(tx, "deleteChildExecutionInfoMap", "", err)
        }
        return nil
}

func updateRequestCancelInfos(
        ctx context.Context,
        tx sqlplugin.Tx,
        requestCancelInfos []*persistence.RequestCancelInfo,
        deleteInfos []int64,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) error {

        if len(requestCancelInfos) > 0 {
                rows := make([]sqlplugin.RequestCancelInfoMapsRow, len(requestCancelInfos))
                for i, requestCancelInfo := range requestCancelInfos {
                        blob, err := parser.RequestCancelInfoToBlob(&serialization.RequestCancelInfo{
                                Version:               requestCancelInfo.Version,
                                InitiatedEventBatchID: requestCancelInfo.InitiatedEventBatchID,
                                CancelRequestID:       requestCancelInfo.CancelRequestID,
                        })
                        if err != nil {
                                return err
                        }
                        rows[i] = sqlplugin.RequestCancelInfoMapsRow{
                                ShardID:      int64(shardID),
                                DomainID:     domainID,
                                WorkflowID:   workflowID,
                                RunID:        runID,
                                InitiatedID:  requestCancelInfo.InitiatedID,
                                Data:         blob.Data,
                                DataEncoding: string(blob.Encoding),
                        }
                }

                if _, err := tx.ReplaceIntoRequestCancelInfoMaps(ctx, rows); err != nil {
                        return convertCommonErrors(tx, "updateRequestCancelInfos", "Failed to execute update query.", err)
                }
        }

        if len(deleteInfos) > 0 {
                if _, err := tx.DeleteFromRequestCancelInfoMaps(ctx, &sqlplugin.RequestCancelInfoMapsFilter{
                        ShardID:      int64(shardID),
                        DomainID:     domainID,
                        WorkflowID:   workflowID,
                        RunID:        runID,
                        InitiatedIDs: deleteInfos,
                }); err != nil {
                        return convertCommonErrors(tx, "updateRequestCancelInfos", "Failed to execute delete query.", err)
                }
        }

        return nil
}

func getRequestCancelInfoMap(
        ctx context.Context,
        db sqlplugin.DB,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) (map[int64]*persistence.RequestCancelInfo, error) {

        rows, err := db.SelectFromRequestCancelInfoMaps(ctx, &sqlplugin.RequestCancelInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        })
        if err != nil && err != sql.ErrNoRows {
                return nil, convertCommonErrors(db, "getRequestCancelInfoMap", "", err)
        }

        ret := make(map[int64]*persistence.RequestCancelInfo)
        for _, row := range rows {
                rowInfo, err := parser.RequestCancelInfoFromBlob(row.Data, row.DataEncoding)
                if err != nil {
                        return nil, err
                }
                ret[row.InitiatedID] = &persistence.RequestCancelInfo{
                        Version:               rowInfo.GetVersion(),
                        InitiatedID:           row.InitiatedID,
                        InitiatedEventBatchID: rowInfo.GetInitiatedEventBatchID(),
                        CancelRequestID:       rowInfo.GetCancelRequestID(),
                }
        }

        return ret, nil
}

func deleteRequestCancelInfoMap(
        ctx context.Context,
        tx sqlplugin.Tx,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
) error {

        if _, err := tx.DeleteFromRequestCancelInfoMaps(ctx, &sqlplugin.RequestCancelInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        }); err != nil {
                return convertCommonErrors(tx, "deleteRequestCancelInfoMap", "", err)
        }
        return nil
}

func updateSignalInfos(
        ctx context.Context,
        tx sqlplugin.Tx,
        signalInfos []*persistence.SignalInfo,
        deleteInfos []int64,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) error {

        if len(signalInfos) > 0 {
                rows := make([]sqlplugin.SignalInfoMapsRow, len(signalInfos))
                for i, signalInfo := range signalInfos {
                        blob, err := parser.SignalInfoToBlob(&serialization.SignalInfo{
                                Version:               signalInfo.Version,
                                InitiatedEventBatchID: signalInfo.InitiatedEventBatchID,
                                RequestID:             signalInfo.SignalRequestID,
                                Name:                  signalInfo.SignalName,
                                Input:                 signalInfo.Input,
                                Control:               signalInfo.Control,
                        })
                        if err != nil {
                                return err
                        }
                        rows[i] = sqlplugin.SignalInfoMapsRow{
                                ShardID:      int64(shardID),
                                DomainID:     domainID,
                                WorkflowID:   workflowID,
                                RunID:        runID,
                                InitiatedID:  signalInfo.InitiatedID,
                                Data:         blob.Data,
                                DataEncoding: string(blob.Encoding),
                        }
                }

                if _, err := tx.ReplaceIntoSignalInfoMaps(ctx, rows); err != nil {
                        return convertCommonErrors(tx, "updateSignalInfos", "Failed to execute update query.", err)
                }
        }

        if len(deleteInfos) > 0 {
                if _, err := tx.DeleteFromSignalInfoMaps(ctx, &sqlplugin.SignalInfoMapsFilter{
                        ShardID:      int64(shardID),
                        DomainID:     domainID,
                        WorkflowID:   workflowID,
                        RunID:        runID,
                        InitiatedIDs: deleteInfos,
                }); err != nil {
                        return convertCommonErrors(tx, "updateSignalInfos", "Failed to execute delete query.", err)
                }
        }

        return nil
}

func getSignalInfoMap(
        ctx context.Context,
        db sqlplugin.DB,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
        parser serialization.Parser,
) (map[int64]*persistence.SignalInfo, error) {

        rows, err := db.SelectFromSignalInfoMaps(ctx, &sqlplugin.SignalInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        })
        if err != nil && err != sql.ErrNoRows {
                return nil, convertCommonErrors(db, "getSignalInfoMap", "", err)
        }

        ret := make(map[int64]*persistence.SignalInfo)
        for _, row := range rows {
                rowInfo, err := parser.SignalInfoFromBlob(row.Data, row.DataEncoding)
                if err != nil {
                        return nil, err
                }
                ret[row.InitiatedID] = &persistence.SignalInfo{
                        Version:               rowInfo.GetVersion(),
                        InitiatedID:           row.InitiatedID,
                        InitiatedEventBatchID: rowInfo.GetInitiatedEventBatchID(),
                        SignalRequestID:       rowInfo.GetRequestID(),
                        SignalName:            rowInfo.GetName(),
                        Input:                 rowInfo.GetInput(),
                        Control:               rowInfo.GetControl(),
                }
        }

        return ret, nil
}

func deleteSignalInfoMap(
        ctx context.Context,
        tx sqlplugin.Tx,
        shardID int,
        domainID serialization.UUID,
        workflowID string,
        runID serialization.UUID,
) error {

        if _, err := tx.DeleteFromSignalInfoMaps(ctx, &sqlplugin.SignalInfoMapsFilter{
                ShardID:    int64(shardID),
                DomainID:   domainID,
                WorkflowID: workflowID,
                RunID:      runID,
        }); err != nil {
                return convertCommonErrors(tx, "deleteSignalInfoMap", "", err)
        }
        return nil
}

uber / cadence / 018dfb1b-c20b-4820-815a-b6aed230920a

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous