Vihren Atlas

namespace_replication_queue.go ×7

Code frontier

unlabeled · c_c5e4a5ac822c

7 tests · 4062 LOC · 177 files · introduces 0 tests · 74 LOC · 5 files

Introduces — evidence that enters the hierarchy at this concept

Code
16 ranges74 lines · 5 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
722 ranges4062 lines · 177 files · Browse complete extent
All tests (intent)
7 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.

5 files ranked by introduced lines: 74 introduced LOC across 16 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/common/persistence/persistence-tests/queue_persistence.go 19 introduced LOC · 1 range

Open complete file

190
191 // TestNamespaceDLQMetadataOperations tests queue metadata operations
192 > func (s *QueuePersistenceSuite) TestNamespaceDLQMetadataOperations() { queue_persistence.go
193 > ackLevel, err := s.GetNamespaceDLQAckLevel(s.ctx)
194 > s.Require().NoError(err)
195 > s.Equal(persistence.EmptyQueueMessageID, ackLevel)
196 >
197 > err = s.UpdateNamespaceDLQAckLevel(s.ctx, 10)
198 > s.NoError(err)
199 >
200 > ackLevel, err = s.GetNamespaceDLQAckLevel(s.ctx)
201 > s.Require().NoError(err)
202 > s.Equal(int64(10), ackLevel)
203 >
204 > err = s.UpdateNamespaceDLQAckLevel(s.ctx, 1)
205 > s.NoError(err)
206 >
207 > ackLevel, err = s.GetNamespaceDLQAckLevel(s.ctx)
208 > s.Require().NoError(err)
209 > s.Equal(int64(10), ackLevel)
210 > }
go.temporal.io/server/common/persistence/persistence_metric_clients.go 18 introduced LOC · 2 ranges

Open complete file

1192 ctx context.Context,
1193 metadata *InternalQueueMetadata,
1194 > ) (retErr error) { persistence_metric_clients.go
1195 > caller := headers.GetCallerInfo(ctx).CallerName
1196 > startTime := time.Now().UTC()
1197 > defer func() {
1198 > p.healthSignals.Record(CallerSegmentMissing, time.Since(startTime), retErr)
1199 > p.recordRequestMetrics(metrics.PersistenceUpdateDLQAckLevelScope, caller, time.Since(startTime), retErr)
1200 > p.recordDataLossMetrics(metrics.PersistenceUpdateDLQAckLevelScope, caller, retErr, "", "")
1201 > }()
1202 > return p.persistence.UpdateDLQAckLevel(ctx, metadata)
1203 }
1204
1205 func (p *queuePersistenceClient) GetDLQAckLevels(
1206 ctx context.Context,
1207 > ) (_ *InternalQueueMetadata, retErr error) { persistence_metric_clients.go
1208 > caller := headers.GetCallerInfo(ctx).CallerName
1209 > startTime := time.Now().UTC()
1210 > defer func() {
1211 > p.healthSignals.Record(CallerSegmentMissing, time.Since(startTime), retErr)
1212 > p.recordRequestMetrics(metrics.PersistenceGetDLQAckLevelScope, caller, time.Since(startTime), retErr)
1213 > p.recordDataLossMetrics(metrics.PersistenceGetDLQAckLevelScope, caller, retErr, "", "")
1214 > }()
1215 > return p.persistence.GetDLQAckLevels(ctx)
1216 }
1217
go.temporal.io/server/common/persistence/namespace_replication_queue.go 17 introduced LOC · 7 ranges

Open complete file

189 var internalMetadata *InternalQueueMetadata
190 if isDLQ {
191 > internalMetadata, ackLevelErr = q.queue.GetDLQAckLevels(ctx) namespace_replication_queue.go
192 } else {
193 internalMetadata, ackLevelErr = q.queue.GetAckLevels(ctx)
205 // Ignore possibly delayed message
206 if ack, ok := ackLevels[clusterName]; ok && ack >= lastProcessedMessageID {
207 > return nil namespace_replication_queue.go
208 > }
209
210 // update ack level
219 internalMetadata.Blob = blob
220 if isDLQ {
221 > err = q.queue.UpdateDLQAckLevel(ctx, internalMetadata) namespace_replication_queue.go
222 } else {
223 err = q.queue.UpdateAckLevel(ctx, internalMetadata)
287 ctx context.Context,
288 lastProcessedMessageID int64,
289 > ) error { namespace_replication_queue.go
290 > return q.updateAckLevelWithRetry(ctx, lastProcessedMessageID, localNamespaceReplicationCluster, true)
291 > }
292
293 func (q *namespaceReplicationQueueImpl) GetDLQAckLevel(
294 ctx context.Context,
295 > ) (int64, error) { namespace_replication_queue.go
296 > metadata, err := q.queue.GetDLQAckLevels(ctx)
297 > if err != nil {
298 return EmptyQueueMessageID, err
299 }
300 > dlqMetadata, err := q.ackLevelsFromBlob(metadata.Blob) namespace_replication_queue.go
301 > if err != nil {
302 return EmptyQueueMessageID, err
303 }
304
305 > ackLevel, ok := dlqMetadata[localNamespaceReplicationCluster] namespace_replication_queue.go
306 > if !ok {
307 > return EmptyQueueMessageID, nil
308 > }
309 > return ackLevel, nil
310 }
311
go.temporal.io/server/common/persistence/persistence_retryable_clients.go 14 introduced LOC · 4 ranges

Open complete file

1165 ctx context.Context,
1166 metadata *InternalQueueMetadata,
1167 > ) error { persistence_retryable_clients.go
1168 > op := func(ctx context.Context) error {
1169 > return p.persistence.UpdateDLQAckLevel(ctx, metadata)
1170 > }
1171
1172 > return backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable) persistence_retryable_clients.go
1173 }
1174
1175 func (p *queueRetryablePersistenceClient) GetDLQAckLevels(
1176 ctx context.Context,
1177 > ) (*InternalQueueMetadata, error) { persistence_retryable_clients.go
1178 > var response *InternalQueueMetadata
1179 > op := func(ctx context.Context) error {
1180 > var err error
1181 > response, err = p.persistence.GetDLQAckLevels(ctx)
1182 > return err
1183 > }
1184
1185 > err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable) persistence_retryable_clients.go
1186 > return response, err
1187 }
1188
go.temporal.io/server/common/persistence/persistence-tests/persistence_test_base.go 6 introduced LOC · 2 ranges

Open complete file

409 ctx context.Context,
410 lastProcessedMessageID int64,
411 > ) error { persistence_test_base.go
412 > return s.NamespaceReplicationQueue.UpdateDLQAckLevel(ctx, lastProcessedMessageID)
413 > }
414
415 // GetNamespaceDLQAckLevel returns namespace dlq ack level
416 func (s *TestBase) GetNamespaceDLQAckLevel(
417 ctx context.Context,
418 > ) (int64, error) { persistence_test_base.go
419 > return s.NamespaceReplicationQueue.GetDLQAckLevel(ctx)
420 > }
421
422 // DeleteMessageFromNamespaceDLQ deletes one message from namespace DLQ

Direct hierarchy

Parents

Children