go.temporal.io/server/common/util.go
791 LOC · 378 covered · 413 uncovered · 110 ranges · 18071 concepts · 73 introducers · 8970 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package common
2
3
import (
4
"context"
5
"errors"
6
"fmt"
7
"math/rand"
8
"reflect"
9
"strings"
10
"sync"
11
"time"
12
13
"github.com/dgryski/go-farm"
14
commonpb "go.temporal.io/api/common/v1"
15
enumspb "go.temporal.io/api/enums/v1"
16
"go.temporal.io/api/serviceerror"
17
"go.temporal.io/api/workflowservice/v1"
18
"go.temporal.io/server/api/historyservice/v1"
19
"go.temporal.io/server/api/matchingservice/v1"
20
workflowspb "go.temporal.io/server/api/workflow/v1"
21
"go.temporal.io/server/common/backoff"
22
"go.temporal.io/server/common/log"
23
"go.temporal.io/server/common/log/tag"
24
"go.temporal.io/server/common/metrics"
25
"go.temporal.io/server/common/primitives/timestamp"
26
serviceerrors "go.temporal.io/server/common/serviceerror"
27
"google.golang.org/grpc/codes"
28
"google.golang.org/grpc/status"
29
"google.golang.org/protobuf/encoding/prototext"
30
"google.golang.org/protobuf/proto"
31
"google.golang.org/protobuf/reflect/protopath"
32
"google.golang.org/protobuf/reflect/protorange"
33
"google.golang.org/protobuf/reflect/protoreflect"
34
"google.golang.org/protobuf/types/known/durationpb"
35
"google.golang.org/protobuf/types/known/timestamppb"
36
)
37
38
const (
39
persistenceClientRetryInitialInterval = 50 * time.Millisecond
40
persistenceClientRetryMaxAttempts = 2
41
42
frontendClientRetryInitialInterval = 200 * time.Millisecond
43
frontendClientRetryMaxAttempts = 2
44
45
historyClientRetryInitialInterval = 50 * time.Millisecond
46
historyClientRetryMaxAttempts = 2
47
48
matchingClientRetryInitialInterval = 1000 * time.Millisecond
49
matchingClientRetryMaxAttempts = 2
50
51
frontendHandlerRetryInitialInterval = 200 * time.Millisecond
52
frontendHandlerRetryMaxInterval = time.Second
53
frontendHandlerRetryMaxAttempts = 2
54
55
historyHandlerRetryInitialInterval = 50 * time.Millisecond
56
historyHandlerRetryMaxAttempts = 2
57
58
matchingHandlerRetryInitialInterval = 1000 * time.Millisecond
59
matchingHandlerRetryMaxAttempts = 2
60
61
readTaskRetryInitialInterval = 50 * time.Millisecond
62
readTaskRetryMaxInterval = 1 * time.Second
63
readTaskRetryExpirationInterval = backoff.NoInterval
64
65
completeTaskRetryInitialInterval = 100 * time.Millisecond
66
completeTaskRetryMaxInterval = 1 * time.Second
67
completeTaskRetryMaxAttempts = 10
68
69
taskRescheduleInitialInterval = 1 * time.Second
70
taskRescheduleBackoffCoefficient = 1.1
71
taskRescheduleMaxInterval = 3 * time.Minute
72
73
taskNotReadyRescheduleInitialInterval = 3 * time.Second
74
taskNotReadyRescheduleBackoffCoefficient = 1.5
75
taskNotReadyRescheduleMaxInterval = 3 * time.Minute
76
77
// dependencyTaskNotCompletedRescheduleInitialInterval is lower than the interval the ack level most queues are
78
// updated at, which can lead to tasks being retried more frequently than they should be. If this becomes an issue,
79
// we should consider increasing this interval.
80
dependencyTaskNotCompletedRescheduleInitialInterval = 3 * time.Second
81
dependencyTaskNotCompletedRescheduleBackoffCoefficient = 1.5
82
dependencyTaskNotCompletedRescheduleMaxInterval = 3 * time.Minute
83
84
taskResourceExhaustedRescheduleInitialInterval = 3 * time.Second
85
taskResourceExhaustedRescheduleBackoffCoefficient = 1.5
86
taskResourceExhaustedRescheduleMaxInterval = 5 * time.Minute
87
88
sdkClientFactoryRetryInitialInterval = 200 * time.Millisecond
89
sdkClientFactoryRetryMaxInterval = 5 * time.Second
90
sdkClientFactoryRetryExpirationInterval = time.Minute
91
92
contextExpireThreshold = 10 * time.Millisecond
93
94
// FailureReasonActivityTimeout is failureReason for when an activity times out, with %v as the timeout type.
95
FailureReasonActivityTimeout = "activity %v timeout"
96
// FailureReasonActivityRetryScheduleToCloseTimeout is failureReason for when an activity retry cannot be scheduled before its schedule-to-close timeout.
97
FailureReasonActivityRetryScheduleToCloseTimeout = "Not enough time to schedule next retry before activity ScheduleToClose timeout, giving up retrying"
98
// FailureReasonCompleteResultExceedsLimit is failureReason for complete result exceeds limit
99
FailureReasonCompleteResultExceedsLimit = "Complete result exceeds size limit."
100
// FailureReasonFailureDetailsExceedsLimit is failureReason for failure details exceeds limit
101
FailureReasonFailureExceedsLimit = "Failure exceeds size limit."
102
// FailureReasonCancelDetailsExceedsLimit is failureReason for cancel details exceeds limit
103
FailureReasonCancelDetailsExceedsLimit = "Cancel details exceed size limit."
104
// FailureReasonHeartbeatExceedsLimit is failureReason for heartbeat exceeds limit
105
FailureReasonHeartbeatExceedsLimit = "Heartbeat details exceed size limit."
106
// FailureReasonHistorySizeExceedsLimit is reason to fail workflow when history size exceeds limit
107
FailureReasonHistorySizeExceedsLimit = "Workflow history size exceeds limit."
108
// FailureReasonHistorySizeExceedsLimit is reason to fail workflow when history count exceeds limit
109
FailureReasonHistoryCountExceedsLimit = "Workflow history count exceeds limit."
110
// FailureReasonMutableStateSizeExceedsLimit is reason to fail workflow when mutable state size exceeds limit
111
FailureReasonMutableStateSizeExceedsLimit = "Workflow mutable state size exceeds limit."
112
// FailureReasonTransactionSizeExceedsLimit is the failureReason for when transaction cannot be committed because it exceeds size limit
113
FailureReasonTransactionSizeExceedsLimit = "Transaction size exceeds limit."
114
// FailureReasonWorkflowTerminationDueToVersionConflict is the failureReason for when workflow is terminated due to version conflict
115
FailureReasonWorkflowTerminationDueToVersionConflict = "Terminate Workflow Due To Version Conflict."
116
)
117
118
var (
119
// ErrBlobSizeExceedsLimit is error for event blob size exceeds limit
120
ErrBlobSizeExceedsLimit = serviceerror.NewInvalidArgument("Blob data size exceeds limit.")
121
// ErrMemoSizeExceedsLimit is error for memo size exceeds limit
122
ErrMemoSizeExceedsLimit = serviceerror.NewInvalidArgument("Memo size exceeds limit.")
123
// ErrContextTimeoutTooShort is error for setting a very short context timeout when calling a long poll API
124
ErrContextTimeoutTooShort = serviceerror.NewFailedPrecondition("Context timeout is too short.")
125
// ErrContextTimeoutNotSet is error for not setting a context timeout when calling a long poll API
126
ErrContextTimeoutNotSet = serviceerror.NewInvalidArgument("Context timeout is not set.")
127
)
128
129
var (
130
// ErrNamespaceHandover is error indicating namespace is in handover state and cannot process request.
131
ErrNamespaceHandover = serviceerror.NewUnavailablef("Namespace replication in %s state.", enumspb.REPLICATION_STATE_HANDOVER)
132
)
133
134
// AwaitWaitGroup calls Wait on the given wait
135
// Returns true if the Wait() call succeeded before the timeout
136
// Returns false if the Wait() did not return before the timeout
138
>
return BlockWithTimeout(wg.Wait, timeout)
139
>
}
140
141
// BlockWithTimeout invokes fn and waits for it to complete until the timeout.
142
// Returns true if the call completed before the timeout, otherwise returns false.
143
// fn is expected to be a blocking call and will continue to occupy a goroutine until it finally completes.
145
>
doneC := make(chan struct{})
146
>
147
>
go func() {
148
>
fn()
149
>
close(doneC)
150
>
}()
151
153
>
defer timer.Stop()
154
>
select {
155
>
case <-doneC:
156
>
return true
157
case <-timer.C:
158
return false
159
}
160
}
161
162
// CreatePersistenceClientRetryPolicy creates a retry policy for calls to persistence
164
>
return backoff.NewExponentialRetryPolicy(persistenceClientRetryInitialInterval).
165
>
WithMaximumAttempts(persistenceClientRetryMaxAttempts)
166
>
}
167
168
// CreateFrontendClientRetryPolicy creates a retry policy for calls to frontend service
170
>
return backoff.NewExponentialRetryPolicy(frontendClientRetryInitialInterval).
171
>
WithMaximumAttempts(frontendClientRetryMaxAttempts)
172
>
}
173
174
// CreateHistoryClientRetryPolicy creates a retry policy for calls to history service.
175
// When retryUnboundedOnSystemResourceExhausted returns true, system-scoped ResourceExhausted
176
// errors retry past the historyClientRetryMaxAttempts cap, bounded only by the policy's
177
// default 1-minute expiration interval and the caller's context. Other errors (and all
178
// errors when the flag is off) follow the standard cap.
179
>
func CreateHistoryClientRetryPolicy(retryUnboundedOnSystemResourceExhausted func() bool) backoff.RetryPolicy {
util.go ×1
180
>
return newClientRetryPolicy(historyClientRetryInitialInterval, historyClientRetryMaxAttempts, retryUnboundedOnSystemResourceExhausted)
181
>
}
182
183
// CreateMatchingClientRetryPolicy creates a retry policy for calls to matching service.
184
// When retryUnboundedOnSystemResourceExhausted returns true, system-scoped ResourceExhausted
185
// errors retry past the matchingClientRetryMaxAttempts cap, bounded only by the policy's
186
// default 1-minute expiration interval and the caller's context. Other errors (and all
187
// errors when the flag is off) follow the standard cap.
188
>
func CreateMatchingClientRetryPolicy(retryUnboundedOnSystemResourceExhausted func() bool) backoff.RetryPolicy {
util.go ×1
189
>
return newClientRetryPolicy(matchingClientRetryInitialInterval, matchingClientRetryMaxAttempts, retryUnboundedOnSystemResourceExhausted)
190
>
}
191
192
>
func newClientRetryPolicy(initialInterval time.Duration, maxAttempts int, retryUnboundedOnSystemResourceExhausted func() bool) backoff.RetryPolicy {
util.go ×1
193
>
capped := backoff.NewExponentialRetryPolicy(initialInterval).
194
>
WithMaximumAttempts(maxAttempts)
195
>
// No max-attempts cap; bounded by the default 1-minute expiration interval
196
>
// and the caller's context.
197
>
extended := backoff.NewExponentialRetryPolicy(initialInterval)
198
>
predicate := func(err error) bool {
199
>
return retryUnboundedOnSystemResourceExhausted() && isSystemResourceExhausted(err)
200
>
}
201
>
return backoff.NewConditionalRetryPolicy(predicate, extended, capped)
202
}
203
205
>
if re, ok := err.(*serviceerror.ResourceExhausted); ok {
207
>
}
209
}
210
211
// CreateMatchingClientLongPollRetryPolicy creates a retry policy for poll calls to matching service
213
>
// no maximum attempts, using default expiration interval of 1 minute
214
>
return backoff.NewExponentialRetryPolicy(matchingClientRetryInitialInterval)
215
>
}
216
217
// CreateFrontendHandlerRetryPolicy creates a retry policy for calls to frontend service
219
>
return backoff.NewExponentialRetryPolicy(frontendHandlerRetryInitialInterval).
220
>
WithMaximumInterval(frontendHandlerRetryMaxInterval).
221
>
WithMaximumAttempts(frontendHandlerRetryMaxAttempts)
222
>
}
223
224
// CreateHistoryHandlerRetryPolicy creates a retry policy for calls to history service
226
>
return backoff.NewExponentialRetryPolicy(historyHandlerRetryInitialInterval).
227
>
WithMaximumAttempts(historyHandlerRetryMaxAttempts)
228
>
}
229
230
// CreateMatchingHandlerRetryPolicy creates a retry policy for calls to matching service
232
>
return backoff.NewExponentialRetryPolicy(matchingHandlerRetryInitialInterval).
233
>
WithMaximumAttempts(matchingHandlerRetryMaxAttempts)
234
>
}
235
236
// CreateReadTaskRetryPolicy creates a retry policy for loading background tasks
238
>
return backoff.NewExponentialRetryPolicy(readTaskRetryInitialInterval).
239
>
WithMaximumInterval(readTaskRetryMaxInterval).
240
>
WithExpirationInterval(readTaskRetryExpirationInterval)
241
>
}
242
243
// CreateCompleteTaskRetryPolicy creates a retry policy for completing background tasks
244
func CreateCompleteTaskRetryPolicy() backoff.RetryPolicy {
245
return backoff.NewExponentialRetryPolicy(completeTaskRetryInitialInterval).
246
WithMaximumInterval(completeTaskRetryMaxInterval).
247
WithMaximumAttempts(completeTaskRetryMaxAttempts)
248
}
249
250
// CreateTaskReschedulePolicy creates a retry policy for rescheduling task with errors not equal to ErrTaskRetry
252
>
return backoff.NewExponentialRetryPolicy(taskRescheduleInitialInterval).
253
>
WithBackoffCoefficient(taskRescheduleBackoffCoefficient).
254
>
WithMaximumInterval(taskRescheduleMaxInterval).
255
>
WithExpirationInterval(backoff.NoInterval)
256
>
}
257
258
// CreateDependencyTaskNotCompletedReschedulePolicy creates a retry policy for rescheduling task with
259
// ErrDependencyTaskNotCompleted
261
>
return backoff.NewExponentialRetryPolicy(dependencyTaskNotCompletedRescheduleInitialInterval).
262
>
WithBackoffCoefficient(dependencyTaskNotCompletedRescheduleBackoffCoefficient).
263
>
WithMaximumInterval(dependencyTaskNotCompletedRescheduleMaxInterval).
264
>
WithExpirationInterval(backoff.NoInterval)
265
>
}
266
267
// CreateTaskNotReadyReschedulePolicy creates a retry policy for rescheduling task with ErrTaskRetry
269
>
return backoff.NewExponentialRetryPolicy(taskNotReadyRescheduleInitialInterval).
270
>
WithBackoffCoefficient(taskNotReadyRescheduleBackoffCoefficient).
271
>
WithMaximumInterval(taskNotReadyRescheduleMaxInterval).
272
>
WithExpirationInterval(backoff.NoInterval)
273
>
}
274
275
// CreateTaskResourceExhaustedReschedulePolicy creates a retry policy for rescheduling task with resource exhausted error
277
>
return backoff.NewExponentialRetryPolicy(taskResourceExhaustedRescheduleInitialInterval).
278
>
WithBackoffCoefficient(taskResourceExhaustedRescheduleBackoffCoefficient).
279
>
WithMaximumInterval(taskResourceExhaustedRescheduleMaxInterval).
280
>
WithExpirationInterval(backoff.NoInterval)
281
>
}
282
283
// CreateSdkClientFactoryRetryPolicy creates a retry policy to handle SdkClientFactory NewClient when frontend service is not ready
285
>
return backoff.NewExponentialRetryPolicy(sdkClientFactoryRetryInitialInterval).
286
>
WithMaximumInterval(sdkClientFactoryRetryMaxInterval).
287
>
WithExpirationInterval(sdkClientFactoryRetryExpirationInterval)
288
>
}
289
290
// IsPersistenceTransientError checks if the error is a transient persistence error
292
>
switch err.(type) {
293
case *serviceerror.Unavailable,
295
>
return true
296
}
297
299
}
300
301
// IsServiceTransientError checks if the error is a retryable error.
303
>
switch err.(type) {
304
case *serviceerror.NotFound,
305
*serviceerror.NamespaceNotFound,
306
*serviceerror.InvalidArgument,
307
*serviceerror.NamespaceNotActive,
309
>
return false
310
}
311
314
>
}
315
317
return false
318
}
319
321
}
322
323
// IsContextDeadlineExceededErr checks if the error is context.DeadlineExceeded or serviceerror.DeadlineExceeded error
325
>
var deadlineExceededSvcErr *serviceerror.DeadlineExceeded
326
>
return errors.Is(err, context.DeadlineExceeded) ||
327
>
errors.As(err, &deadlineExceededSvcErr)
328
>
}
329
330
// IsContextCanceledErr checks if the error is context.Canceled or serviceerror.Canceled error
332
>
var canceledSvcErr *serviceerror.Canceled
333
>
return errors.Is(err, context.Canceled) ||
334
>
errors.As(err, &canceledSvcErr)
335
>
}
336
337
// IsServiceClientTransientError checks if the error is a transient error.
339
>
if IsServiceHandlerRetryableError(err) {
341
>
}
342
345
>
}
346
348
case *serviceerrors.ShardOwnershipLost,
349
*serviceerrors.StalePartitionCounts:
350
return true
351
}
352
354
}
355
357
>
if IsNamespaceHandoverError(err) {
359
>
}
360
362
case *serviceerror.Internal,
364
>
return true
366
>
for _, opErr := range err.OperationErrors() {
367
>
if opErr != nil && IsServiceHandlerRetryableError(opErr) {
369
>
}
370
}
371
}
372
374
}
375
377
>
return err.Error() == ErrNamespaceHandover.Error()
378
>
}
379
380
func IsStickyWorkerUnavailable(err error) bool {
381
switch err.(type) {
382
case *serviceerrors.StickyWorkerUnavailable:
383
return true
384
}
385
return false
386
}
387
388
// IsResourceExhausted checks if the error is a service busy error.
390
>
switch err.(type) {
392
>
return true
393
}
395
}
396
397
// IsInternalError checks if the error is an internal error.
399
>
var internalErr *serviceerror.Internal
400
>
return errors.As(err, &internalErr)
401
>
}
402
403
// IsNotFoundError checks if the error is a not found error.
405
>
var notFoundErr *serviceerror.NotFound
406
>
return errors.As(err, ¬FoundErr)
407
>
}
408
410
>
if err != nil {
411
>
return fmt.Sprintf("%08x", farm.Fingerprint32([]byte(err.Error())))
412
>
}
413
return "00000000"
414
}
415
416
// WorkflowIDToHistoryShard is used to map namespaceID-workflowID pair to a shardID.
417
// TODO: rename to BusinessIDToHistoryShard.
418
func WorkflowIDToHistoryShard(
419
namespaceID string,
420
workflowID string,
421
numberOfShards int32,
423
>
idBytes := []byte(namespaceID + "_" + workflowID)
424
>
hash := farm.Fingerprint32(idBytes)
425
>
return int32(hash%uint32(numberOfShards)) + 1 // ShardID starts with 1
426
>
}
427
428
func MapShardID(
429
sourceShardCount int32,
430
targetShardCount int32,
431
sourceShardID int32,
433
>
if sourceShardCount%targetShardCount != 0 && targetShardCount%sourceShardCount != 0 {
434
panic(fmt.Sprintf("cannot map shard ID between source & target shard count: %v vs %v",
435
sourceShardCount, targetShardCount))
436
}
437
439
>
if sourceShardCount < targetShardCount {
441
>
// 0-3
442
>
// 0-15
443
>
// 0 -> 0, 4, 8, 12
444
>
// 1 -> 1, 5, 9, 13
445
>
// 2 -> 2, 6, 10, 14
446
>
// 3 -> 3, 7, 11, 15
447
>
// 4x
448
>
ratio := targetShardCount / sourceShardCount
449
>
targetShardIDs := make([]int32, ratio)
450
>
for i := range targetShardIDs {
451
>
targetShardIDs[i] = sourceShardID + int32(i)*sourceShardCount + 1
452
>
}
453
>
return targetShardIDs
455
>
// many to one
456
>
return []int32{(sourceShardID % targetShardCount) + 1}
457
>
} else {
458
return []int32{sourceShardID + 1}
459
}
460
}
461
462
func VerifyShardIDMapping(
463
thisShardCount int32,
464
thatShardCount int32,
465
thisShardID int32,
466
thatShardID int32,
468
>
if thisShardCount%thatShardCount != 0 && thatShardCount%thisShardCount != 0 {
469
panic(fmt.Sprintf("cannot verify shard ID mapping between diff shard count: %v vs %v",
470
thisShardCount, thatShardCount))
471
}
473
>
if thisShardID%shardCountMin == thatShardID%shardCountMin {
474
>
return nil
475
>
}
477
>
"shard ID mapping verification failed; shard count: %v vs %v, shard ID: %v vs %v",
478
>
thisShardCount, thatShardCount,
479
>
thisShardID, thatShardID,
480
>
)
481
}
482
483
func PrettyPrint[T proto.Message](msgs []T, header ...string) {
484
var sb strings.Builder
485
_, _ = sb.WriteString("==========================================================================\n")
486
for _, h := range header {
487
_, _ = sb.WriteString(h)
488
_, _ = sb.WriteRune('\n')
489
}
490
_, _ = sb.WriteString("--------------------------------------------------------------------------\n")
491
for _, m := range msgs {
492
bs, _ := prototext.Marshal(m)
493
sb.Write(bs)
494
sb.WriteRune('\n')
495
}
496
fmt.Print(sb.String())
497
}
498
499
// IsValidContext checks that the thrift context is not expired on cancelled.
500
// Returns nil if the context is still valid. Otherwise, returns the result of
501
// ctx.Err()
503
>
ch := ctx.Done()
504
>
if ch != nil {
507
>
return ctx.Err()
509
>
return nil
510
}
511
}
513
>
if ok && time.Until(deadline) < contextExpireThreshold {
514
return context.DeadlineExceeded
515
}
517
}
518
519
// GenerateRandomString is used for generate test string
521
>
letterRunes := []rune("random")
522
>
b := make([]rune, n)
523
>
for i := range b {
524
>
b[i] = letterRunes[rand.Intn(len(letterRunes))]
525
>
}
526
>
return string(b)
527
}
528
529
// CreateMatchingPollWorkflowTaskQueueResponse create response for matching's PollWorkflowTaskQueue
530
>
func CreateMatchingPollWorkflowTaskQueueResponse(historyResponse *historyservice.RecordWorkflowTaskStartedResponse, workflowExecution *commonpb.WorkflowExecution, token []byte) *matchingservice.PollWorkflowTaskQueueResponseWithRawHistory {
request_response.pb.go ×6
531
>
matchingResp := &matchingservice.PollWorkflowTaskQueueResponseWithRawHistory{
532
>
TaskToken: token,
533
>
WorkflowExecution: workflowExecution,
534
>
WorkflowType: historyResponse.WorkflowType,
535
>
PreviousStartedEventId: historyResponse.PreviousStartedEventId,
536
>
StartedEventId: historyResponse.StartedEventId,
537
>
Attempt: historyResponse.GetAttempt(),
538
>
NextEventId: historyResponse.NextEventId,
539
>
StickyExecutionEnabled: historyResponse.StickyExecutionEnabled,
540
>
TransientWorkflowTask: historyResponse.TransientWorkflowTask,
541
>
WorkflowExecutionTaskQueue: historyResponse.WorkflowExecutionTaskQueue,
542
>
BranchToken: historyResponse.BranchToken,
543
>
ScheduledTime: historyResponse.ScheduledTime,
544
>
StartedTime: historyResponse.StartedTime,
545
>
Queries: historyResponse.Queries,
546
>
Messages: historyResponse.Messages,
547
>
History: historyResponse.History,
548
>
NextPageToken: historyResponse.NextPageToken,
549
>
RawHistory: historyResponse.RawHistoryBytes,
550
>
}
551
>
552
>
return matchingResp
553
>
}
554
555
// CreateHistoryStartWorkflowRequest create a start workflow request for history.
556
// Assumes startRequest is valid. See frontend workflow_handler for detailed validation logic.
557
func CreateHistoryStartWorkflowRequest(
558
namespaceID string,
559
startRequest *workflowservice.StartWorkflowExecutionRequest,
560
parentExecutionInfo *workflowspb.ParentExecutionInfo,
561
rootExecutionInfo *workflowspb.RootExecutionInfo,
562
now time.Time,
564
>
// We include the original startRequest in the forwarded request to History, but
565
>
// we don't want to send workflow payloads twice. We deep copy to a new struct,
566
>
// rather than mutate the request, to accommodate internal retries.
567
>
if startRequest.ContinuedFailure != nil || startRequest.LastCompletionResult != nil {
569
>
}
571
>
NamespaceId: namespaceID,
572
>
StartRequest: startRequest,
573
>
ContinueAsNewInitiator: enumspb.CONTINUE_AS_NEW_INITIATOR_UNSPECIFIED,
574
>
Attempt: 1,
575
>
ParentExecutionInfo: parentExecutionInfo,
576
>
FirstWorkflowTaskBackoff: durationpb.New(backoff.GetBackoffForNextScheduleNonNegative(startRequest.GetCronSchedule(), now, now)),
577
>
ContinuedFailure: startRequest.ContinuedFailure,
578
>
LastCompletionResult: startRequest.LastCompletionResult,
579
>
RootExecutionInfo: rootExecutionInfo,
580
>
VersioningOverride: startRequest.GetVersioningOverride(),
581
>
}
582
>
startRequest.ContinuedFailure = nil
583
>
startRequest.LastCompletionResult = nil
584
>
585
>
if timestamp.DurationValue(startRequest.GetWorkflowExecutionTimeout()) > 0 {
586
>
deadline := now.Add(timestamp.DurationValue(startRequest.GetWorkflowExecutionTimeout()))
util.go ×1
587
>
histRequest.WorkflowExecutionExpirationTime = timestamppb.New(deadline.Round(time.Millisecond))
588
>
}
589
590
// CronSchedule and WorkflowStartDelay should not both be set on the same request
592
>
histRequest.ContinueAsNewInitiator = enumspb.CONTINUE_AS_NEW_INITIATOR_CRON_SCHEDULE
handler.go ×25
593
>
}
594
596
histRequest.FirstWorkflowTaskBackoff = startRequest.GetWorkflowStartDelay()
597
}
598
600
}
601
602
// CheckEventBlobSizeLimit checks if a blob data exceeds limits. It logs a warning if it exceeds warnLimit,
603
// and return ErrBlobSizeExceedsLimit if it exceeds errorLimit.
604
func CheckEventBlobSizeLimit(
605
actualSize int,
606
warnLimit int,
607
errorLimit int,
608
namespace string,
609
workflowID string,
610
runID string,
611
metricsHandler metrics.Handler,
612
logger log.Logger,
613
operation string,
615
>
616
>
metrics.EventBlobSize.With(metricsHandler).Record(int64(actualSize), metrics.OperationTag(operation))
617
>
if actualSize > warnLimit {
619
>
logger.Warn("Blob data size exceeds the warning limit.",
620
>
tag.WorkflowNamespace(namespace), // TODO: Not necessarily a "workflow" namespace, fix the tag.
621
>
tag.WorkflowID(workflowID), // TODO: this should be entity ID and we need an archetype too.
622
>
tag.WorkflowRunID(runID), // TODO: not necessarily a workflow run ID, fix the tag.
623
>
tag.WorkflowSize(int64(actualSize)),
624
>
tag.BlobSizeViolationOperation(operation))
625
>
}
626
628
>
metrics.BlobSizeError.With(metricsHandler).Record(1, metrics.OperationTag(operation))
util.go ×1
629
>
return ErrBlobSizeExceedsLimit
630
>
}
631
}
633
}
634
635
// ValidateLongPollContextTimeout checks if the context timeout for a long poll handler is too short or below a normal value.
636
// If the timeout is not set or too short, it logs an error, and returns ErrContextTimeoutNotSet or ErrContextTimeoutTooShort
637
// accordingly. If the timeout is only below a normal value, it just logs an info and returns nil.
638
func ValidateLongPollContextTimeout(
639
ctx context.Context,
640
handlerName string,
641
logger log.Logger,
643
>
644
>
deadline, err := ValidateLongPollContextTimeoutIsSet(ctx, handlerName, logger)
645
>
if err != nil {
647
>
}
649
>
if timeout < MinLongPollTimeout {
651
>
logger.Error("Context timeout is too short for long poll API.",
652
>
tag.WorkflowHandlerName(handlerName), tag.Error(err), tag.WorkflowPollContextTimeout(timeout))
653
>
return err
654
>
}
656
logger.Warn("Context timeout is lower than critical value for long poll API.",
657
tag.WorkflowHandlerName(handlerName), tag.WorkflowPollContextTimeout(timeout))
658
}
660
}
661
662
// ValidateLongPollContextTimeoutIsSet checks if the context timeout is set for long poll requests.
663
func ValidateLongPollContextTimeoutIsSet(
664
ctx context.Context,
665
handlerName string,
666
logger log.Logger,
668
>
669
>
deadline, ok := ctx.Deadline()
670
>
if !ok {
672
>
logger.Error("Context timeout not set for long poll API.",
673
>
tag.WorkflowHandlerName(handlerName), tag.Error(err))
674
>
return deadline, err
675
>
}
677
}
678
679
>
func GetPayloadsMapSize(data map[string]*commonpb.Payloads) int {
command_attr_validator.go ×4
680
>
size := 0
681
>
for key, payloads := range data {
682
>
size += len(key)
683
>
size += payloads.Size()
684
>
}
685
687
}
688
689
// CloneProto is a generic typed version of proto.Clone from proto.
691
>
return proto.Clone(v).(T)
692
>
}
693
695
>
if src == nil {
696
return nil
697
}
698
700
>
for k, v := range src {
701
>
result[k] = CloneProto(v)
702
>
}
703
>
return result
704
}
705
706
// DiscardUnknownProto discards unknown fields in a proto message.
708
>
return protorange.Range(m.ProtoReflect(), func(values protopath.Values) error {
709
>
m, ok := values.Index(-1).Value.Interface().(protoreflect.Message)
710
>
if ok && len(m.GetUnknown()) > 0 {
712
>
}
714
})
715
}
716
717
// MergeProtoExcludingFields merges fields from source into target, excluding specific fields.
718
// The fields to exclude are specified as pointers to fields in the target struct.
719
>
func MergeProtoExcludingFields(target, source proto.Message, doNotSyncFunc func(v any) []any) error {
util.go ×7
720
>
if target == nil || source == nil {
721
return serviceerror.NewInvalidArgument("target and source cannot be nil")
722
}
723
725
>
return serviceerror.NewInvalidArgument("target and source must be of the same type")
util.go ×1
726
>
}
727
729
>
excludeSet := make(map[string]struct{}, len(excludeFields))
730
>
for _, fieldPtr := range excludeFields {
731
>
fieldName, err := getFieldNameFromStruct(target, fieldPtr)
732
>
if err != nil {
733
return err
734
}
736
}
737
739
>
dstVal := reflect.ValueOf(target).Elem()
740
>
for i := 0; i < srcVal.NumField(); i++ {
741
>
field := srcVal.Type().Field(i)
742
>
if _, exclude := excludeSet[field.Name]; !exclude {
743
>
srcField := srcVal.Field(i)
744
>
dstField := dstVal.Field(i)
745
>
if dstField.CanSet() {
746
>
dstField.Set(srcField)
747
>
}
748
}
749
}
750
752
}
753
755
>
structVal := reflect.ValueOf(structPtr).Elem()
756
>
for i := 0; i < structVal.NumField(); i++ {
757
>
field := structVal.Field(i)
758
>
if field.CanSet() && field.Addr().Interface() == fieldPtr {
759
>
return structVal.Type().Field(i).Name, nil
760
>
}
761
}
762
return "", serviceerror.NewInternal("field not found in the struct")
763
}
764
765
// IsRetryableRPCError checks if the error is a retryable gRPC error.
767
>
var st *status.Status
768
>
stGetter, ok := err.(interface{ Status() *status.Status })
769
>
if ok {
770
>
st = stGetter.Status()
771
>
} else {
772
st, ok = status.FromError(err)
773
if !ok {
774
// Not a gRPC induced error
775
return false
776
}
777
}
778
// nolint:exhaustive
780
case codes.Canceled,
781
codes.Unknown,
782
codes.Unavailable,
783
codes.DeadlineExceeded,
784
codes.ResourceExhausted,
785
codes.Aborted,
787
>
return true
789
>
return false
790
}
791
}