go.temporal.io/server/service/matching/db.go
995 LOC · 594 covered · 401 uncovered · 137 ranges · 1162 concepts · 82 introducers · 509 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 matching
2
3
import (
4
"cmp"
5
"context"
6
"errors"
7
"fmt"
8
"math"
9
"slices"
10
"sync"
11
"time"
12
13
enumspb "go.temporal.io/api/enums/v1"
14
"go.temporal.io/api/serviceerror"
15
persistencespb "go.temporal.io/server/api/persistence/v1"
16
"go.temporal.io/server/common/log"
17
"go.temporal.io/server/common/log/tag"
18
"go.temporal.io/server/common/metrics"
19
"go.temporal.io/server/common/persistence"
20
"go.temporal.io/server/common/primitives/timestamp"
21
"go.temporal.io/server/common/softassert"
22
"go.temporal.io/server/service/matching/counter"
23
"google.golang.org/protobuf/proto"
24
"google.golang.org/protobuf/types/known/timestamppb"
25
)
26
27
const (
28
initialRangeID = 1 // Id of the first range of a new task queue
29
30
// Subqueue zero corresponds to "the queue" before migrating metadata to subqueues.
31
// For backwards compatibility, some operations only apply to subqueue zero for now.
32
subqueueZero = subqueueIndex(0)
33
)
34
35
type (
36
taskQueueDB struct {
37
// constants
38
config *taskQueueConfig
39
queue *PhysicalTaskQueueKey
40
isDraining bool
41
store persistence.TaskManager
42
logger log.Logger
43
metricsHandler metrics.Handler
44
45
// mutable
46
sync.Mutex
47
rangeID int64
48
subqueues []*dbSubqueue
49
otherHasTasks bool
50
scaleState *persistencespb.PartitionScaleState
51
52
// used to avoid unnecessary metadata writes:
53
lastChange time.Time // updated when metadata is changed in memory
54
lastWrite time.Time // updated when metadata is successfully written to db
55
}
56
57
dbSubqueue struct {
58
persistencespb.SubqueueInfo
59
maxReadLevel int64
60
oldestTime time.Time // time of oldest task if backlog, otherwise zero time
61
}
62
63
taskQueueState struct {
64
rangeID int64
65
ackLevel int64 // TODO(pri): old matcher cleanup, delete later
66
subqueues []persistencespb.SubqueueInfo
67
otherHasTasks bool
68
scaleState *persistencespb.PartitionScaleState
69
}
70
71
subqueueIndex int
72
73
createTasksResponse struct {
74
bySubqueue map[subqueueIndex]subqueueCreateTasksResponse
75
}
76
77
subqueueCreateTasksResponse struct {
78
tasks []*persistencespb.AllocatedTaskInfo
79
maxReadLevelBefore int64
80
maxReadLevelAfter int64
81
}
82
83
createFairTasksResponse map[subqueueIndex][]*persistencespb.AllocatedTaskInfo // subqueue -> tasks
84
)
85
86
// newTaskQueueDB returns an instance of an object that represents
87
// persistence view of a physical task queue. All mutations / reads to queues
88
// wrt persistence go through this object.
89
//
90
// This class will serialize writes to persistence that do condition updates. There are
91
// two reasons for doing this:
92
// - To work around known Cassandra issue where concurrent LWT to the same partition cause timeout errors
93
// - To provide the guarantee that there is only writer who updates queue in persistence at any given point in time
94
// This guarantee makes some of the other code simpler and there is no impact to perf because updates to taskqueue are
95
// spread out and happen in background routines
96
func newTaskQueueDB(
97
config *taskQueueConfig,
98
store persistence.TaskManager,
99
queue *PhysicalTaskQueueKey,
100
logger log.Logger,
101
metricsHandler metrics.Handler,
102
isDraining bool,
104
>
return &taskQueueDB{
105
>
config: config,
106
>
queue: queue,
107
>
isDraining: isDraining,
108
>
store: store,
109
>
logger: logger,
110
>
metricsHandler: metricsHandler,
111
>
}
112
>
}
113
114
// RangeID returns the current persistence view of rangeID
116
>
db.Lock()
117
>
defer db.Unlock()
118
>
return db.rangeID
119
>
}
120
121
// GetMaxReadLevel returns the current maxReadLevel
123
>
db.Lock()
124
>
defer db.Unlock()
125
>
return db.getMaxReadLevelLocked(subqueue)
126
>
}
127
129
>
return db.subqueues[subqueue].maxReadLevel
130
>
}
131
132
// GetMaxReadLevel returns the current maxReadLevel
133
func (db *taskQueueDB) GetMaxFairReadLevel(subqueue subqueueIndex) fairLevel {
134
db.Lock()
135
defer db.Unlock()
136
return db.getMaxFairReadLevelLocked(subqueue)
137
}
138
139
func (db *taskQueueDB) getMaxFairReadLevelLocked(subqueue subqueueIndex) fairLevel {
140
return fairLevelFromProto(db.subqueues[subqueue].FairMaxReadLevel)
141
}
142
143
// This is only exposed for testing!
144
>
func (db *taskQueueDB) setMaxReadLevelForTesting(subqueue subqueueIndex, level int64) {
db.go ×1
145
>
db.Lock()
146
>
defer db.Unlock()
147
>
db.subqueues[subqueue].maxReadLevel = level
148
>
}
149
150
// RenewLease renews the lease on a taskqueue. If there is no previous lease,
151
// this method will attempt to steal taskqueue from current owner
152
func (db *taskQueueDB) RenewLease(
153
ctx context.Context,
155
>
db.Lock()
156
>
defer db.Unlock()
157
>
158
>
if db.rangeID == 0 {
159
>
if err := db.takeOverTaskQueueLocked(ctx); err != nil {
161
>
}
163
>
if err := db.updateTaskQueueLocked(ctx, true); err != nil {
165
>
}
166
}
168
>
rangeID: db.rangeID,
169
>
ackLevel: db.subqueues[subqueueZero].AckLevel, // TODO(pri): cleanup, only used by old backlog manager
170
>
subqueues: db.cloneSubqueues(),
171
>
otherHasTasks: !db.isDraining && db.otherHasTasks,
172
>
scaleState: db.scaleState,
173
>
}, nil
174
}
175
176
func (db *taskQueueDB) takeOverTaskQueueLocked(
177
ctx context.Context,
179
>
response, err := db.store.GetTaskQueue(ctx, &persistence.GetTaskQueueRequest{
180
>
NamespaceID: db.queue.NamespaceId(),
181
>
TaskQueue: db.queue.PersistenceName(),
182
>
TaskType: db.queue.TaskType(),
183
>
})
184
>
switch err.(type) {
186
>
db.rangeID = response.RangeID
187
>
// If we are the draining one, then assume the other has tasks, so we can migrate
188
>
// backwards safely.
189
>
db.otherHasTasks = response.TaskQueueInfo.OtherHasTasks || db.isDraining
190
>
db.subqueues = db.ensureDefaultSubqueuesLocked(
191
>
response.TaskQueueInfo.Subqueues,
192
>
response.TaskQueueInfo.AckLevel,
193
>
response.TaskQueueInfo.ApproximateBacklogCount,
194
>
)
195
>
db.scaleState = response.TaskQueueInfo.PartitionScaleState
196
>
err := db.updateTaskQueueLocked(ctx, true)
197
>
if err != nil {
199
>
return err
200
>
}
202
>
// We took over the task queue and are not sure what tasks may have been written
203
>
// before. Set max read level id of all subqueues to just before our new block.
204
>
maxReadLevel := rangeIDToTaskIDBlock(db.rangeID, db.config.RangeSize).start - 1
205
>
for _, s := range db.subqueues {
206
>
s.maxReadLevel = maxReadLevel
207
>
}
208
>
return nil
209
211
>
db.rangeID = initialRangeID
212
>
db.subqueues = db.ensureDefaultSubqueuesLocked(nil, 0, 0)
213
>
214
>
// If we are the draining one, then assume the other has tasks, so we can migrate
215
>
// backwards safely. Also assume other has tasks if the config allows for migration
216
>
// (and the partition supports fairness) since we may have just turned on fairness and need to migrate.
217
>
canMigrate := (db.config.NewMatcher || db.config.EnableFairness) && db.queue.Partition().SupportsFairness()
218
>
db.otherHasTasks = canMigrate || db.isDraining
219
>
220
>
if _, err := db.store.CreateTaskQueue(ctx, &persistence.CreateTaskQueueRequest{
221
>
RangeID: db.rangeID,
222
>
TaskQueueInfo: db.cachedQueueInfo(),
223
>
}); err != nil {
225
>
return err
226
>
}
228
>
// In this case, ensureDefaultSubqueuesLocked already initialized subqueue 0 to have
229
>
// ackLevel and maxReadLevel 0, so we don't need to initialize them.
230
>
softassert.That(db.logger, db.subqueues[0].maxReadLevel == 0, "should have maxReadLevel 0 here")
231
>
softassert.That(db.logger, db.subqueues[0].FairMaxReadLevel == nil, "should have maxReadLevel 0 here")
232
>
softassert.That(db.logger, db.subqueues[0].AckLevel == 0, "should have ackLevel 0 here")
233
>
softassert.That(db.logger, db.subqueues[0].FairAckLevel == nil, "should have ackLevel 0 here")
234
>
return nil
235
237
>
return err
238
}
239
}
240
241
>
func (db *taskQueueDB) updateTaskQueueLocked(ctx context.Context, incrementRangeId bool) error {
db.go ×2
242
>
newRangeID := db.rangeID
243
>
if incrementRangeId {
245
>
}
247
>
RangeID: newRangeID,
248
>
TaskQueueInfo: db.cachedQueueInfo(),
249
>
PrevRangeID: db.rangeID,
250
>
}); err != nil {
252
>
}
254
>
db.rangeID = newRangeID
255
>
return nil
256
}
257
258
// OldUpdateState updates the queue state with the given value. This is used by old backlog
259
// manager (not subqueue-enabled).
260
// TODO(pri): old matcher cleanup
261
func (db *taskQueueDB) OldUpdateState(
262
ctx context.Context,
263
ackLevel int64,
265
>
db.Lock()
266
>
defer db.Unlock()
267
>
// We don't need to update lastWrite/lastChange in here since this function is only used by
268
>
// the old backlog manager and those fields are only used by the new backlog manager.
269
>
270
>
// Reset approximateBacklogCount to fix the count divergence issue
271
>
maxReadLevel := db.getMaxReadLevelLocked(subqueueZero)
272
>
if ackLevel == maxReadLevel {
274
>
db.subqueues[subqueueZero].oldestTime = time.Time{} // zero time means no backlog
275
>
}
276
278
>
db.subqueues[subqueueZero].AckLevel = ackLevel
279
>
280
>
err := db.updateTaskQueueLocked(ctx, false)
281
>
if err != nil {
283
>
}
285
>
return err
286
}
287
288
// shouldUpdateMetadataOnAppendLocked returns whether a task append should also write the
289
// metadata blob. This is always true when enough time has passed since the last metadata
290
// write (controlled by MetadataUpdateOnAppendInterval), so that backlog counts stay
291
// reasonably fresh. When the interval is zero, metadata is updated on every append
292
// (previous behavior). Caller must hold db.Mutex.
294
>
interval := db.config.MetadataUpdateOnAppendInterval()
295
>
return interval <= 0 || time.Since(db.lastWrite) >= interval
296
>
}
297
299
>
db.Lock()
300
>
defer db.Unlock()
301
>
defer db.emitPhysicalBacklogGaugesLocked()
302
>
303
>
// We only need to write if something changed, or if we're past half of the persistence TTL.
304
>
// Cap at 24h so that the scavenger (which looks for metadata not updated in 48h) doesn't
305
>
// mistake the queue for idle, even if a future partition kind has a longer TTL.
306
>
ttl := min(24*time.Hour, cmp.Or(db.queue.Partition().PersistenceTTL(), 24*time.Hour))
307
>
needWrite := db.lastChange.After(db.lastWrite) || time.Since(db.lastWrite) > ttl/2
308
>
if !needWrite {
310
>
// momentarily (this could happen due to eventual consistency of membership updates).
311
>
// So instead, do a (cheaper) read to just check the range id.
312
>
return db.verifyOwnershipLocked(ctx)
313
>
}
314
316
}
317
319
>
response, err := db.store.GetTaskQueue(ctx, &persistence.GetTaskQueueRequest{
320
>
NamespaceID: db.queue.NamespaceId(),
321
>
TaskQueue: db.queue.PersistenceName(),
322
>
TaskType: db.queue.TaskType(),
323
>
})
324
>
if err != nil {
325
return err
326
}
329
>
Msg: fmt.Sprintf("task queue ownership lost: stored rangeID %d, in-memory rangeID %d",
330
>
response.RangeID, db.rangeID),
331
>
}
332
>
}
334
}
335
336
>
func (db *taskQueueDB) updateAckLevelAndBacklogStats(subqueue subqueueIndex, newAckLevel int64, countDelta int64, oldestTime time.Time) {
db.go ×3
337
>
db.Lock()
338
>
defer db.Unlock()
339
>
340
>
dbQueue := db.subqueues[subqueue]
341
>
if newAckLevel < dbQueue.AckLevel {
342
softassert.Fail(db.logger,
343
"ack level in subqueue should not move backwards",
344
tag.Int("subqueue-id", int(subqueue)),
345
tag.Any("cur-ack-level", dbQueue.AckLevel),
346
tag.Any("new-ack-level", newAckLevel))
347
}
350
>
dbQueue.AckLevel = newAckLevel
351
>
}
352
355
>
if dbQueue.ApproximateBacklogCount != 0 || !dbQueue.oldestTime.Equal(oldestTime) {
357
>
dbQueue.ApproximateBacklogCount = 0
358
>
dbQueue.oldestTime = oldestTime
359
>
}
362
>
db.updateBacklogStatsLocked(subqueue, countDelta, oldestTime)
363
>
}
364
}
365
366
>
func (db *taskQueueDB) updateFairAckLevel(subqueue subqueueIndex, newAckLevel fairLevel, countDelta, knownCount int64, oldestTime time.Time) {
db.go ×3
367
>
db.Lock()
368
>
defer db.Unlock()
369
>
370
>
db.lastChange = time.Now()
371
>
dbQueue := db.subqueues[subqueue]
372
>
if prev := fairLevelFromProto(dbQueue.FairAckLevel); newAckLevel.less(prev) {
373
softassert.Fail(db.logger,
374
"ack level in subqueue should not move backwards",
375
tag.Int("subqueue-id", int(subqueue)),
376
tag.Any("cur-ack-level", prev),
377
tag.Any("new-ack-level", newAckLevel))
378
}
380
>
381
>
if knownCount >= 0 {
383
>
dbQueue.ApproximateBacklogCount = knownCount
384
>
dbQueue.oldestTime = oldestTime
387
>
}
388
}
389
390
// Use this to reset ApproximateBacklogCount when the backlog count is known, e.g. when you're
391
// read to the end of the backlog.
392
>
func (db *taskQueueDB) setKnownFairBacklogCount(subqueue subqueueIndex, count int64) {
fair_task_reader.go ×33
393
>
db.Lock()
394
>
defer db.Unlock()
395
>
396
>
if db.subqueues[subqueue].ApproximateBacklogCount != count {
398
>
db.subqueues[subqueue].ApproximateBacklogCount = count
399
>
if count == 0 {
401
>
}
402
}
403
}
404
405
// updateApproximateBacklogCount updates the in-memory DB state with the given delta value
406
// TODO(pri): old matcher cleanup
408
>
db.Lock()
409
>
defer db.Unlock()
410
>
db.lastChange = time.Now()
411
>
db.updateBacklogStatsLocked(subqueueZero, countDelta, oldestTime)
412
>
}
413
414
>
func (db *taskQueueDB) updateBacklogStatsLocked(subqueue subqueueIndex, countDelta int64, oldestTime time.Time) {
db.go ×3
415
>
// Prevent under-counting
416
>
count := &db.subqueues[subqueue].ApproximateBacklogCount
417
>
if *count+countDelta < 0 {
419
>
tag.WorkerVersion(db.queue.Version().MetricsTagValue()),
420
>
tag.WorkflowNamespaceID(db.queue.Partition().NamespaceId()))
421
>
*count = 0
424
>
}
426
}
427
428
>
func (db *taskQueueDB) persistTopKFairnessKeys(subqueue subqueueIndex, entries []counter.TopKEntry) {
db.go ×2
429
>
db.Lock()
430
>
defer db.Unlock()
431
>
432
>
counts := make([]*persistencespb.FairnessKeyCount, len(entries))
433
>
for i, entry := range entries {
434
>
counts[i] = &persistencespb.FairnessKeyCount{Key: entry.Key, Count: entry.Count}
435
>
}
436
438
>
db.lastChange = time.Now()
439
}
440
441
>
func (db *taskQueueDB) getTopKFairnessKeys(subqueue subqueueIndex) []counter.TopKEntry {
fair_task_writer.go ×17
442
>
db.Lock()
443
>
defer db.Unlock()
444
>
445
>
if subqueue >= subqueueIndex(len(db.subqueues)) {
446
return nil
447
}
449
>
entries := make([]counter.TopKEntry, len(counts))
450
>
for i, count := range counts {
452
>
}
454
}
455
456
// getApproximateBacklogCountsBySubqueue return the approximate backlog count for each subqueue.
457
// The index corresponds to the subqueue id.
459
>
db.Lock()
460
>
defer db.Unlock()
461
>
462
>
result := make([]int64, len(db.subqueues))
463
>
for id, s := range db.subqueues {
464
>
result[id] = s.ApproximateBacklogCount
465
>
}
466
>
return result
467
}
468
469
>
func (db *taskQueueDB) getApproximateBacklogCountAndMaxReadLevel(subqueue subqueueIndex) (int64, fairLevel) {
message.pb.go ×1
470
>
db.Lock()
471
>
defer db.Unlock()
472
>
s := db.subqueues[subqueue]
473
>
return s.ApproximateBacklogCount, fairLevelFromProto(s.FairMaxReadLevel)
474
>
}
475
477
>
db.Lock()
478
>
defer db.Unlock()
479
>
480
>
var total int64
481
>
for _, s := range db.subqueues {
482
>
total += s.ApproximateBacklogCount
483
>
}
484
>
return total
485
}
486
487
// SetOtherHasTasks updates the otherHasTasks flag and attempts to persist immediately.
488
// The in-memory state is updated regardless of whether persistence succeeds.
489
// Returns any error from the persistence attempt.
490
>
func (db *taskQueueDB) SetOtherHasTasks(ctx context.Context, value bool) error {
physical_task_queue_manager.go ×3
491
>
db.Lock()
492
>
defer db.Unlock()
493
>
if db.otherHasTasks == value {
494
return nil
495
}
497
>
db.lastChange = time.Now()
498
>
return db.updateTaskQueueLocked(ctx, false)
499
}
500
501
// UpdateScaleState sets the partition scale state (in memory). If syncToDB is true, it also tries to persist it to the DB.
502
// If syncToDB is false, ctx is not used.
503
func (db *taskQueueDB) UpdateScaleState(ctx context.Context, scaleState *persistencespb.PartitionScaleState, syncToDB bool) error {
504
db.Lock()
505
defer db.Unlock()
506
db.scaleState = scaleState
507
db.lastChange = time.Now()
508
if syncToDB {
509
return db.updateTaskQueueLocked(ctx, false)
510
}
511
return nil
512
}
513
514
// CreateTasks creates a batch of given tasks for this task queue
515
func (db *taskQueueDB) CreateTasks(
516
ctx context.Context,
517
reqs []*writeTaskRequest,
519
>
if db.isDraining {
520
return createTasksResponse{}, softassert.UnexpectedInternalErr(db.logger, "CreateTasks can't be used in draining mode", nil)
521
}
522
524
>
defer db.Unlock()
525
>
526
>
if len(reqs) == 0 {
527
return createTasksResponse{}, nil
528
}
529
531
>
allTasks := make([]*persistencespb.AllocatedTaskInfo, len(reqs))
532
>
allSubqueues := make([]int, len(reqs))
533
>
for i, req := range reqs {
534
>
task := &persistencespb.AllocatedTaskInfo{
535
>
TaskId: req.id,
536
>
Data: req.taskInfo,
537
>
}
538
>
allTasks[i] = task
539
>
allSubqueues[i] = int(req.subqueue)
540
>
541
>
u := updates[req.subqueue]
542
>
updates[req.subqueue] = subqueueCreateTasksResponse{
543
>
tasks: append(u.tasks, task),
544
>
maxReadLevelBefore: db.getMaxReadLevelLocked(req.subqueue),
545
>
maxReadLevelAfter: task.TaskId, // task ids are in order so this is the max
546
>
}
547
>
}
548
550
>
db.subqueues[sq].ApproximateBacklogCount += int64(len(update.tasks))
551
>
}
552
553
// Decide whether to include metadata in the write. We always need the LWT for the
554
// range ID check, but updating the full metadata blob on every append has extra cost.
555
// We piggyback the metadata update if enough time has passed since the last write.
557
>
558
>
resp, err := db.store.CreateTasks(
559
>
ctx,
560
>
&persistence.CreateTasksRequest{
561
>
TaskQueueInfo: &persistence.PersistedTaskQueueInfo{
562
>
Data: db.cachedQueueInfo(),
563
>
RangeID: db.rangeID,
564
>
},
565
>
Tasks: allTasks,
566
>
Subqueues: allSubqueues,
567
>
UpdateMetadata: updateMetadata,
568
>
})
569
>
570
>
// Update the maxReadLevel after the writes are completed, but before we send the response,
571
>
// so that taskReader is guaranteed to see the new read level when SpoolTask wakes it up.
572
>
// Do this even if the write fails, we won't reuse the task ids.
573
>
for sq, update := range updates {
574
>
db.subqueues[sq].maxReadLevel = update.maxReadLevelAfter
575
>
}
576
578
>
// Only update lastWrite for persistence implementations that update metadata on CreateTasks,
db.go ×2
579
>
// otherwise we have a change to ApproximateBacklogCount we need to write.
580
>
if resp.UpdatedMetadata {
584
>
}
586
>
// tasks definitely were not created, restore the counter. For other errors tasks may or may not be created.
db.go ×1
587
>
// In those cases we keep the count incremented, hence it may be an overestimate.
588
>
for i, update := range updates {
589
>
db.subqueues[i].ApproximateBacklogCount -= int64(len(update.tasks))
590
>
}
591
}
593
}
594
595
// CreateFairTasks creates a batch of given tasks for this task queue
596
func (db *taskQueueDB) CreateFairTasks(
597
ctx context.Context,
598
reqs []*writeTaskRequest,
600
>
if db.isDraining {
601
return createFairTasksResponse{}, softassert.UnexpectedInternalErr(db.logger, "CreateTasks can't be used in draining mode", nil)
602
}
603
605
>
defer db.Unlock()
606
>
607
>
if len(reqs) == 0 {
608
return nil, nil
609
}
610
612
>
newMaxLevel := make(map[subqueueIndex]fairLevel)
613
>
allTasks := make([]*persistencespb.AllocatedTaskInfo, len(reqs))
614
>
allSubqueues := make([]int, len(reqs))
615
>
for i, req := range reqs {
616
>
task := &persistencespb.AllocatedTaskInfo{
617
>
TaskId: req.id,
618
>
TaskPass: req.pass,
619
>
Data: req.taskInfo,
620
>
}
621
>
allTasks[i] = task
622
>
allSubqueues[i] = int(req.subqueue)
623
>
newTasks[req.subqueue] = append(newTasks[req.subqueue], task)
624
>
newMaxLevel[req.subqueue] = newMaxLevel[req.subqueue].max(req.fairLevel)
625
>
}
626
628
>
db.subqueues[sq].ApproximateBacklogCount += int64(len(tasks))
629
>
}
630
631
// Unlike in CreateTasks, we can set the persisted FairMaxReadLevel before persisting.
632
// This means that for stores that update metadata along with writing tasks (i.e. Cassandra),
633
// the FairMaxReadLevel will be more up-to-date. The max read level is not used by
634
// fairTaskReader, so there's no correctness issue with doing this.
636
>
db.subqueues[sq].FairMaxReadLevel = fairLevelFromProto(db.subqueues[sq].FairMaxReadLevel).max(level).toProto()
637
>
}
638
640
>
641
>
resp, err := db.store.CreateTasks(
642
>
ctx,
643
>
&persistence.CreateTasksRequest{
644
>
TaskQueueInfo: &persistence.PersistedTaskQueueInfo{
645
>
Data: db.cachedQueueInfo(),
646
>
RangeID: db.rangeID,
647
>
},
648
>
Tasks: allTasks,
649
>
Subqueues: allSubqueues,
650
>
UpdateMetadata: updateMetadata,
651
>
})
652
>
653
>
if err == nil {
654
>
// Only update lastWrite for persistence implementations that update metadata on CreateTasks,
db.go ×2
655
>
// otherwise we have a change to ApproximateBacklogCount we need to write.
656
>
if resp.UpdatedMetadata {
660
>
}
662
>
// Tasks definitely were not created, restore the counter. For other errors tasks may or may not be created.
db.go ×1
663
>
// In those cases we keep the count incremented, hence it may be an overestimate.
664
>
// Don't bother restoring MaxReadLevel, it's okay if that's too high.
665
>
for i, tasks := range newTasks {
666
>
db.subqueues[i].ApproximateBacklogCount -= int64(len(tasks))
667
>
}
668
}
670
}
671
672
// writeDefinitelyFailed returns whether an error from a CreateTasks call indicates the tasks
673
// were definitely not persisted, so that we can safely un-increment ApproximateBacklogCount.
674
// We have to be conservative: for most errors (e.g. Unavailable) the write may or may not have
675
// reached the database, so we leave the count incremented and accept a possible overestimate.
676
// Only errors that reject the write before it reaches the database qualify:
677
// - ConditionFailedError: the range ID LWT failed, so the batch was rejected.
678
// - ResourceExhausted with a persistence rate-limit or concurrent-limit cause: dropped by
679
// the persistence rate limiter (see persistence.ErrPersistence*LimitExceeded).
681
>
if _, ok := err.(*persistence.ConditionFailedError); ok {
683
>
}
686
case enumspb.RESOURCE_EXHAUSTED_CAUSE_PERSISTENCE_LIMIT,
688
>
return true
689
}
690
}
692
}
693
694
// GetTasks returns a batch of tasks between the given range
695
func (db *taskQueueDB) GetTasks(
696
ctx context.Context,
697
subqueue subqueueIndex,
698
inclusiveMinTaskID int64,
699
exclusiveMaxTaskID int64,
700
batchSize int,
702
>
return db.store.GetTasks(ctx, &persistence.GetTasksRequest{
703
>
NamespaceID: db.queue.NamespaceId(),
704
>
TaskQueue: db.queue.PersistenceName(),
705
>
TaskType: db.queue.TaskType(),
706
>
InclusiveMinTaskID: inclusiveMinTaskID,
707
>
ExclusiveMaxTaskID: exclusiveMaxTaskID,
708
>
Subqueue: int(subqueue),
709
>
PageSize: batchSize,
710
>
})
711
>
}
712
713
// GetFairTasks returns a batch of tasks after the given level
714
func (db *taskQueueDB) GetFairTasks(
715
ctx context.Context,
716
subqueue subqueueIndex,
717
inclusiveMinLevel fairLevel,
718
batchSize int,
720
>
return db.store.GetTasks(ctx, &persistence.GetTasksRequest{
721
>
NamespaceID: db.queue.NamespaceId(),
722
>
TaskQueue: db.queue.PersistenceName(),
723
>
TaskType: db.queue.TaskType(),
724
>
InclusiveMinPass: inclusiveMinLevel.pass,
725
>
InclusiveMinTaskID: inclusiveMinLevel.id,
726
>
ExclusiveMaxTaskID: math.MaxInt64,
727
>
Subqueue: int(subqueue),
728
>
PageSize: batchSize,
729
>
UseLimit: true,
730
>
})
731
>
}
732
733
// CompleteTasksLessThan deletes of tasks less than the given taskID. Limit is
734
// the upper bound of number of tasks that can be deleted by this method. It may
735
// or may not be honored
736
func (db *taskQueueDB) CompleteTasksLessThan(
737
ctx context.Context,
738
exclusiveMaxTaskID int64,
739
limit int,
740
subqueue subqueueIndex,
742
>
n, err := db.store.CompleteTasksLessThan(ctx, &persistence.CompleteTasksLessThanRequest{
743
>
NamespaceID: db.queue.NamespaceId(),
744
>
TaskQueueName: db.queue.PersistenceName(),
745
>
TaskType: db.queue.TaskType(),
746
>
ExclusiveMaxTaskID: exclusiveMaxTaskID,
747
>
Subqueue: int(subqueue),
748
>
Limit: limit,
749
>
})
750
>
if err != nil {
751
db.logger.Error("Persistent store operation failure",
752
tag.StoreOperationCompleteTasksLessThan,
753
tag.Error(err),
754
tag.TaskID(exclusiveMaxTaskID),
755
tag.WorkflowTaskQueueType(db.queue.TaskType()),
756
tag.WorkflowTaskQueueName(db.queue.PersistenceName()),
757
)
758
}
760
}
761
762
// CompleteFairTasksLessThan deletes of tasks less than the given taskID. Limit is
763
// the upper bound of number of tasks that can be deleted by this method. It may
764
// or may not be honored
765
func (db *taskQueueDB) CompleteFairTasksLessThan(
766
ctx context.Context,
767
exclusiveMaxLevel fairLevel,
768
limit int,
769
subqueue subqueueIndex,
771
>
n, err := db.store.CompleteTasksLessThan(ctx, &persistence.CompleteTasksLessThanRequest{
772
>
NamespaceID: db.queue.NamespaceId(),
773
>
TaskQueueName: db.queue.PersistenceName(),
774
>
TaskType: db.queue.TaskType(),
775
>
ExclusiveMaxPass: exclusiveMaxLevel.pass,
776
>
ExclusiveMaxTaskID: exclusiveMaxLevel.id,
777
>
Subqueue: int(subqueue),
778
>
Limit: limit,
779
>
})
780
>
if err != nil {
781
db.logger.Error("Persistent store operation failure",
782
tag.StoreOperationCompleteTasksLessThan,
783
tag.Error(err),
784
tag.AckLevel(exclusiveMaxLevel),
785
tag.WorkflowTaskQueueType(db.queue.TaskType()),
786
tag.WorkflowTaskQueueName(db.queue.PersistenceName()),
787
)
788
}
790
}
791
792
func (db *taskQueueDB) AllocateSubqueue(
793
ctx context.Context,
794
key *persistencespb.SubqueueKey,
795
) ([]persistencespb.SubqueueInfo, error) {
796
db.Lock()
797
defer db.Unlock()
798
799
newSubqueue := db.newSubqueueLocked(key)
800
db.subqueues = append(db.subqueues, newSubqueue)
801
802
// ensure written to metadata before returning
803
err := db.updateTaskQueueLocked(ctx, false)
804
if err != nil {
805
// If this was a conflict, caller will shut down partition. Otherwise, we don't know
806
// for sure if this write made it to persistence or not. We should forget about the new
807
// subqueue and let a future call to AllocateSubqueue add it again. If we crash and
808
// reload, the new owner may see the subqueue present, which is also fine.
809
db.subqueues = db.subqueues[:len(db.subqueues)-1]
810
return nil, err
811
}
812
813
return db.cloneSubqueues(), nil
814
}
815
817
>
if ttl := db.queue.Partition().PersistenceTTL(); ttl > 0 {
819
>
}
821
}
822
824
>
infos := make([]*persistencespb.SubqueueInfo, len(db.subqueues))
825
>
for i := range db.subqueues {
826
>
infos[i] = &db.subqueues[i].SubqueueInfo
827
>
}
828
>
return &persistencespb.TaskQueueInfo{
829
>
NamespaceId: db.queue.NamespaceId(),
830
>
Name: db.queue.PersistenceName(),
831
>
TaskType: db.queue.TaskType(),
832
>
Kind: db.queue.Partition().Kind(),
833
>
AckLevel: db.subqueues[subqueueZero].AckLevel, // backwards compatibility
834
>
ExpiryTime: db.expiryTime(),
835
>
LastUpdateTime: timestamp.TimeNowPtrUtc(),
836
>
ApproximateBacklogCount: db.subqueues[subqueueZero].ApproximateBacklogCount, // backwards compatibility
837
>
Subqueues: infos,
838
>
OtherHasTasks: db.otherHasTasks,
839
>
PartitionScaleState: db.scaleState,
840
>
}
841
}
842
843
// emitPhysicalBacklogGaugesLocked emits backlog gauges tagged by priority key, along with
844
// the legacy task_lag_per_tl gauge.
845
//
846
// When version-attributed backlog metrics are enabled (BacklogMetricsEmitInterval > 0), this
847
// emits physical_approximate_backlog_count and physical_approximate_backlog_age_seconds for
848
// the unversioned queue only. Version-attributed metrics (including appropriate attribution of
849
// the default queue's tasks to current and ramping versions) are emitted separately by the
850
// partition manager via fetchAndEmitLogicalBacklogMetrics.
851
//
852
// When version-attributed metrics are disabled (BacklogMetricsEmitInterval == 0), this falls back
853
// to emitting the original approximate_backlog_count and approximate_backlog_age_seconds for
854
// all queues (including versioned queues when BreakdownMetricsByBuildID is enabled).
856
>
if !db.config.BreakdownMetricsByTaskQueue() || !db.config.BreakdownMetricsByPartition() {
857
return
858
}
859
861
>
862
>
if attributionEnabled {
863
>
if db.queue.IsVersioned() {
865
>
}
866
} else {
867
if db.queue.IsVersioned() && !db.config.BreakdownMetricsByBuildID() {
868
return
869
}
870
}
871
873
>
var oldestTime time.Time
874
>
counts := make(map[int32]int64)
875
>
for _, s := range db.subqueues {
876
>
counts[s.Key.Priority] += s.ApproximateBacklogCount
877
>
oldestTime = minNonZeroTime(oldestTime, s.oldestTime)
878
>
// note: this metric is only an estimation for the lag.
879
>
// taskID in DB may not be continuous, especially when task list ownership changes.
880
>
if s.FairMaxReadLevel != nil && s.FairAckLevel != nil {
882
>
// get rid of this metric.
883
>
totalLag += s.FairMaxReadLevel.TaskId - s.FairAckLevel.TaskId
886
>
}
887
}
888
890
>
backlogAgeGauge := metrics.ApproximateBacklogAgeSeconds
891
>
if attributionEnabled {
892
>
backlogCountGauge = metrics.PhysicalApproximateBacklogCount
893
>
backlogAgeGauge = metrics.PhysicalApproximateBacklogAgeSeconds
894
>
}
895
897
>
backlogCountGauge.With(db.metricsHandler).Record(float64(count), metrics.MatchingTaskPriorityTag(priority))
898
>
}
899
>
if oldestTime.IsZero() {
903
>
}
905
}
906
907
func (db *taskQueueDB) ensureDefaultSubqueuesLocked(
908
infos []*persistencespb.SubqueueInfo,
909
initAckLevel int64,
910
initApproxCount int64,
912
>
// convert+copy protos to []*dbSubqueue
913
>
subqueues := make([]*dbSubqueue, len(infos))
914
>
for i, info := range infos {
916
>
proto.Merge(&subqueues[i].SubqueueInfo, info)
917
>
}
918
919
// check for default priority and add if not present (this may be initializing subqueue 0)
921
>
Priority: int32(db.config.DefaultPriorityKey),
922
>
}
923
>
hasDefault := slices.ContainsFunc(subqueues, func(s *dbSubqueue) bool {
925
>
})
927
>
subqueues = append(subqueues, db.newSubqueueLocked(defKey))
928
>
// If we are transitioning from no-subqueues to subqueues, initialize subqueue 0 with
929
>
// the ack level and approx count from TaskQueueInfo.
930
>
if len(subqueues) == 1 {
931
>
subqueues[subqueueZero].AckLevel = initAckLevel
932
>
subqueues[subqueueZero].ApproximateBacklogCount = initApproxCount
933
>
}
934
}
936
}
937
938
>
func (db *taskQueueDB) newSubqueueLocked(key *persistencespb.SubqueueKey) *dbSubqueue {
db.go ×11
939
>
// For fifo queues: start ack level + max read level just before the current block.
940
>
// For fair queues: ack level and max read level don't matter here.
941
>
initAckLevel := rangeIDToTaskIDBlock(db.rangeID, db.config.RangeSize).start - 1
942
>
softassert.That(db.logger, initAckLevel >= 0, "initAckLevel should not be negative")
943
>
944
>
s := &dbSubqueue{maxReadLevel: initAckLevel}
945
>
s.Key = key
946
>
s.AckLevel = initAckLevel
947
>
return s
948
>
}
949
950
// clone db.subqueues so we can return it outside our lock
952
>
infos := make([]persistencespb.SubqueueInfo, len(db.subqueues))
953
>
for i := range db.subqueues {
954
>
proto.Merge(&infos[i], &db.subqueues[i].SubqueueInfo)
955
>
}
956
>
return infos
957
}
958
960
>
if !db.config.BreakdownMetricsByTaskQueue() || !db.config.BreakdownMetricsByPartition() {
961
return
962
}
963
964
>
attributionEnabled := db.config.BacklogMetricsEmitInterval() > 0
physical_task_queue_manager.go ×3
965
>
966
>
if attributionEnabled {
967
>
if db.queue.IsVersioned() {
969
>
}
970
} else {
971
if db.queue.IsVersioned() && !db.config.BreakdownMetricsByBuildID() {
972
return
973
}
974
}
975
977
>
db.Lock()
978
>
for _, s := range db.subqueues {
979
>
priorities[s.Key.Priority] = struct{}{}
980
>
}
981
>
db.Unlock()
982
>
983
>
backlogCountGauge := metrics.ApproximateBacklogCount
984
>
backlogAgeGauge := metrics.ApproximateBacklogAgeSeconds
985
>
if attributionEnabled {
986
>
backlogCountGauge = metrics.PhysicalApproximateBacklogCount
987
>
backlogAgeGauge = metrics.PhysicalApproximateBacklogAgeSeconds
988
>
}
989
991
>
backlogCountGauge.With(db.metricsHandler).Record(0, metrics.MatchingTaskPriorityTag(k))
992
>
}
993
>
backlogAgeGauge.With(db.metricsHandler).Record(0)
994
>
metrics.TaskLagPerTaskQueueGauge.With(db.metricsHandler).Record(0)
995
}