Vihren Atlas

dlq_writer.go ×8

Code frontier

unlabeled · c_2f0d35c530d3

10 tests · 2717 LOC · 136 files · introduces 0 tests · 71 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
11 ranges71 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
420 ranges2717 lines · 136 files · Browse complete extent
All tests (intent)
10 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

3 files ranked by introduced lines: 71 introduced LOC across 11 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/queues/dlq_writer.go 58 introduced LOC · 8 ranges

Open complete file

68 task tasks.Task,
69 isNamespaceActive bool,
70 > ) error { dlq_writer.go
71 > queueKey := persistence.QueueKey{
72 > QueueType: persistence.QueueTypeHistoryDLQ,
73 > Category: task.GetCategory(),
74 > SourceCluster: sourceCluster,
75 > TargetCluster: targetCluster,
76 > }
77 > _, err := q.dlqWriter.CreateQueue(ctx, &persistence.CreateQueueRequest{
78 > QueueKey: queueKey,
79 > })
80 > if err != nil {
81 if !errors.Is(err, persistence.ErrQueueAlreadyExists) {
82 return fmt.Errorf("%w: %v", ErrCreateDLQ, err)
84 }
85
86 > resp, err := func() (*persistence.EnqueueTaskResponse, error) { dlq_writer.go
87 > // Acquire a process-level lock for this specific DLQ to prevent concurrent writes
88 > // from multiple shards causing CAS conflicts in the persistence layer.
89 > mu := q.getQueueMutex(queueKey)
90 > mu.Lock()
91 > defer mu.Unlock()
92 >
93 > return q.dlqWriter.EnqueueTask(ctx, &persistence.EnqueueTaskRequest{
94 > QueueType: queueKey.QueueType,
95 > SourceCluster: queueKey.SourceCluster,
96 > TargetCluster: queueKey.TargetCluster,
97 > Task: task,
98 > SourceShardID: sourceShardID,
99 > })
100 > }()
101 > if err != nil {
102 return fmt.Errorf("%w: %v", ErrSendTaskToDLQ, err)
103 }
104
105 > nsMetricTag := metrics.NamespaceUnknownTag() dlq_writer.go
106 > var nsLogTag tag.Tag
107 > ns, err := q.namespaceRegistry.GetNamespaceByID(namespace.ID(task.GetNamespaceID()))
108 > if err != nil {
109 q.logger.Warn("Failed to get namespace name while trying to write a task to DLQ",
110 tag.WorkflowNamespace(task.GetNamespaceID()),
112 )
113 nsLogTag = tag.WorkflowNamespaceID(task.GetNamespaceID())
114 > } else { dlq_writer.go
115 nsMetricTag = metrics.NamespaceTag(ns.Name().String())
116 nsLogTag = tag.WorkflowNamespace(ns.Name().String())
118
119 // "passive" means the namespace is in standby mode and only replicates data
120 > namespaceState := metrics.PassiveNamespaceStateTagValue dlq_writer.go
121 > if isNamespaceActive {
122 namespaceState = metrics.ActiveNamespaceStateTagValue
123 }
124 > taskType := GetTaskTypeTagValue(task, isNamespaceActive, q.chasmRegistry) dlq_writer.go
125 > metrics.DLQWrites.With(q.metricsHandler).Record(
126 > 1,
127 > metrics.TaskCategoryTag(task.GetCategory().Name()),
128 > metrics.NamespaceStateTag(namespaceState),
129 > nsMetricTag,
130 > metrics.TaskTypeTag(taskType),
131 > metrics.OperationTag(taskType),
132 > getArchetypeTag(task, q.chasmRegistry),
133 > )
134 > q.logger.Warn("Task enqueued to DLQ",
135 > tag.DLQMessageID(resp.Metadata.ID),
136 > tag.SourceCluster(sourceCluster),
137 > tag.TargetCluster(targetCluster),
138 > tag.TaskType(task.GetType()),
139 > tag.String("task-category", task.GetCategory().Name()),
140 > nsLogTag,
141 > )
142 > return nil
143 }
144
145 // getQueueMutex returns a per-queue mutex, creating it if it doesn't exist.
146 // This provides process-level locking to serialize concurrent writes to the same queue.
147 > func (q *DLQWriter) getQueueMutex(queueKey persistence.QueueKey) *sync.Mutex { dlq_writer.go
148 > if mu, ok := q.enqueueMutex.Load(queueKey); ok {
149 return mu.(*sync.Mutex) //nolint:revive
150 }
151
152 > newMutex := &sync.Mutex{} dlq_writer.go
153 > actual, _ := q.enqueueMutex.LoadOrStore(queueKey, newMutex)
154 > return actual.(*sync.Mutex) //nolint:revive
155 }
go.temporal.io/server/service/history/queues/queuestest/fake_queue_writer.go 10 introduced LOC · 2 ranges

Open complete file

28 ctx context.Context,
29 request *persistence.EnqueueTaskRequest,
30 > ) (*persistence.EnqueueTaskResponse, error) { fake_queue_writer.go
31 > // Protect the slice append from concurrent access
32 > d.mu.Lock()
33 > d.EnqueueTaskRequests = append(d.EnqueueTaskRequests, request)
34 > d.mu.Unlock()
35 >
36 > if d.EnqueueTaskFunc != nil {
37 return d.EnqueueTaskFunc(ctx, request)
38 }
43 context.Context,
44 *persistence.CreateQueueRequest,
45 > ) (*persistence.CreateQueueResponse, error) { fake_queue_writer.go
46 > return nil, d.CreateQueueErr
47 > }
go.temporal.io/server/common/log/tag/tags.go 3 introduced LOC · 1 range

Open complete file

667 }
668
669 > func DLQMessageID(dlqMessageID int64) ZapTag { tags.go
670 > return NewInt64("dlq-message-id", dlqMessageID)
671 > }
672
673 // retry

Direct hierarchy

Parents

Children