Vihren Atlas

namespace_replication_queue.go ×5

Code frontier

unlabeled · c_41e453efdaf5

7 tests · 4067 LOC · 177 files · introduces 0 tests · 79 LOC · 5 files

Introduces — evidence that enters the hierarchy at this concept

Code
14 ranges79 lines · 5 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
720 ranges4067 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: 79 introduced LOC across 14 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

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

Open complete file

92
93 // TestQueueMetadataOperations tests queue metadata operations
94 > func (s *QueuePersistenceSuite) TestQueueMetadataOperations() { queue_persistence.go
95 > clusterAckLevels, err := s.GetAckLevels(s.ctx)
96 > s.Require().NoError(err)
97 > s.Assert().Len(clusterAckLevels, 0)
98 >
99 > err = s.UpdateAckLevel(s.ctx, 10, "test1")
100 > s.Require().NoError(err)
101 >
102 > clusterAckLevels, err = s.GetAckLevels(s.ctx)
103 > s.Require().NoError(err)
104 > s.Assert().Len(clusterAckLevels, 1)
105 > s.Assert().Equal(int64(10), clusterAckLevels["test1"])
106 >
107 > err = s.UpdateAckLevel(s.ctx, 20, "test1")
108 > s.Require().NoError(err)
109 >
110 > clusterAckLevels, err = s.GetAckLevels(s.ctx)
111 > s.Require().NoError(err)
112 > s.Assert().Len(clusterAckLevels, 1)
113 > s.Assert().Equal(int64(20), clusterAckLevels["test1"])
114 >
115 > err = s.UpdateAckLevel(s.ctx, 25, "test2")
116 > s.Require().NoError(err)
117 >
118 > clusterAckLevels, err = s.GetAckLevels(s.ctx)
119 > s.Require().NoError(err)
120 > s.Assert().Len(clusterAckLevels, 2)
121 > s.Assert().Equal(int64(20), clusterAckLevels["test1"])
122 > s.Assert().Equal(int64(25), clusterAckLevels["test2"])
123 > }
124
125 // TestNamespaceReplicationDLQ tests namespace DLQ operations
go.temporal.io/server/common/persistence/persistence_metric_clients.go 18 introduced LOC · 2 ranges

Open complete file

1091 ctx context.Context,
1092 metadata *InternalQueueMetadata,
1093 > ) (retErr error) { persistence_metric_clients.go
1094 > caller := headers.GetCallerInfo(ctx).CallerName
1095 > startTime := time.Now().UTC()
1096 > defer func() {
1097 > p.healthSignals.Record(CallerSegmentMissing, time.Since(startTime), retErr)
1098 > p.recordRequestMetrics(metrics.PersistenceUpdateAckLevelScope, caller, time.Since(startTime), retErr)
1099 > p.recordDataLossMetrics(metrics.PersistenceUpdateAckLevelScope, caller, retErr, "", "")
1100 > }()
1101 > return p.persistence.UpdateAckLevel(ctx, metadata)
1102 }
1103
1104 func (p *queuePersistenceClient) GetAckLevels(
1105 ctx context.Context,
1106 > ) (_ *InternalQueueMetadata, retErr error) { persistence_metric_clients.go
1107 > caller := headers.GetCallerInfo(ctx).CallerName
1108 > startTime := time.Now().UTC()
1109 > defer func() {
1110 > p.healthSignals.Record(CallerSegmentMissing, time.Since(startTime), retErr)
1111 > p.recordRequestMetrics(metrics.PersistenceGetAckLevelScope, caller, time.Since(startTime), retErr)
1112 > p.recordDataLossMetrics(metrics.PersistenceGetAckLevelScope, caller, retErr, "", "")
1113 > }()
1114 > return p.persistence.GetAckLevels(ctx)
1115 }
1116
go.temporal.io/server/common/persistence/persistence_retryable_clients.go 14 introduced LOC · 4 ranges

Open complete file

1084 ctx context.Context,
1085 metadata *InternalQueueMetadata,
1086 > ) error { persistence_retryable_clients.go
1087 > op := func(ctx context.Context) error {
1088 > return p.persistence.UpdateAckLevel(ctx, metadata)
1089 > }
1090
1091 > return backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable) persistence_retryable_clients.go
1092 }
1093
1094 func (p *queueRetryablePersistenceClient) GetAckLevels(
1095 ctx context.Context,
1096 > ) (*InternalQueueMetadata, error) { persistence_retryable_clients.go
1097 > var response *InternalQueueMetadata
1098 > op := func(ctx context.Context) error {
1099 > var err error
1100 > response, err = p.persistence.GetAckLevels(ctx)
1101 > return err
1102 > }
1103
1104 > err := backoff.ThrottleRetryContext(ctx, op, p.policy, p.isRetryable) persistence_retryable_clients.go
1105 > return response, err
1106 }
1107
go.temporal.io/server/common/persistence/namespace_replication_queue.go 11 introduced LOC · 5 ranges

Open complete file

145 lastProcessedMessageID int64,
146 clusterName string,
147 > ) error { namespace_replication_queue.go
148 > return q.updateAckLevelWithRetry(ctx, lastProcessedMessageID, clusterName, false)
149 > }
150
151 func (q *namespaceReplicationQueueImpl) DeleteMessagesBefore(
191 internalMetadata, ackLevelErr = q.queue.GetDLQAckLevels(ctx)
192 } else {
193 > internalMetadata, ackLevelErr = q.queue.GetAckLevels(ctx) namespace_replication_queue.go
194 > }
195
196 if ackLevelErr != nil {
221 err = q.queue.UpdateDLQAckLevel(ctx, internalMetadata)
222 } else {
223 > err = q.queue.UpdateAckLevel(ctx, internalMetadata) namespace_replication_queue.go
224 > }
225 if err != nil {
226 return fmt.Errorf("failed to update ack level: %v", err)
232 func (q *namespaceReplicationQueueImpl) GetAckLevels(
233 ctx context.Context,
234 > ) (map[string]int64, error) { namespace_replication_queue.go
235 > metadata, err := q.queue.GetAckLevels(ctx)
236 > if err != nil {
237 return nil, err
238 }
239 > return q.ackLevelsFromBlob(metadata.Blob) namespace_replication_queue.go
240 }
241
go.temporal.io/server/common/persistence/persistence-tests/persistence_test_base.go 6 introduced LOC · 2 ranges

Open complete file

360 lastProcessedMessageID int64,
361 clusterName string,
362 > ) error { persistence_test_base.go
363 > return s.NamespaceReplicationQueue.UpdateAckLevel(ctx, lastProcessedMessageID, clusterName)
364 > }
365
366 // GetAckLevels returns replication queue ack levels
367 func (s *TestBase) GetAckLevels(
368 ctx context.Context,
369 > ) (map[string]int64, error) { persistence_test_base.go
370 > return s.NamespaceReplicationQueue.GetAckLevels(ctx)
371 > }
372
373 // PublishToNamespaceDLQ is a utility method to add messages to the namespace DLQ

Direct hierarchy

Parents

Children