Vihren Atlas

service.go ×8

Code frontier

unlabeled · c_7b21990010cf

11 tests · 19726 LOC · 572 files · introduces 0 tests · 413 LOC · 47 files

Introduces — evidence that enters the hierarchy at this concept

Code
96 ranges413 lines · 47 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
3744 ranges19726 lines · 572 files · Browse complete extent
All tests (intent)
11 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.

Showing the top 20 of 47 files by introduced lines: 311 of 413 introduced LOC and 60 of 96 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/frontend/service.go 38 introduced LOC · 5 ranges

Open complete file

519
520 // Stop stops the service
521 > func (s *Service) Stop() { service.go
522 > // initiate graceful shutdown:
523 > // 1. Fail rpc health check, this will cause client side load balancer to stop forwarding requests to this node
524 > // 2. wait for failure detection time
525 > // 3. stop taking new requests by returning InternalServiceError
526 > // 4. Wait for X second
527 > // 5. Stop everything forcefully and return
528 >
529 > requestDrainTime := max(time.Second, s.config.ShutdownDrainDuration())
530 > failureDetectionTime := max(0, s.config.ShutdownFailHealthCheckDuration())
531 >
532 > s.logger.Info("ShutdownHandler: Updating gRPC health status to ShuttingDown")
533 > s.healthServer.Shutdown()
534 > s.membershipMonitor.SetDraining(true)
535 >
536 > s.logger.Info("ShutdownHandler: Waiting for others to discover I am unhealthy")
537 > time.Sleep(failureDetectionTime)
538 >
539 > s.handler.Stop()
540 > s.operatorHandler.Stop()
541 > s.adminHandler.Stop()
542 > s.versionChecker.Stop()
543 > s.visibilityManager.Close()
544 >
545 > s.logger.Info("ShutdownHandler: Draining traffic")
546 > // Gracefully stop gRPC server and HTTP API server concurrently
547 > var wg sync.WaitGroup
548 > wg.Go(func() {
549 > t := time.AfterFunc(requestDrainTime, func() {
550 s.logger.Info("ShutdownHandler: Drain time expired, stopping all traffic")
551 s.server.Stop()
552 })
553 > s.server.GracefulStop() service.go
554 > t.Stop()
555 })
556 > if s.httpAPIServer != nil { service.go
557 wg.Go(func() {
558 s.httpAPIServer.GracefulStop(requestDrainTime)
559 })
560 }
561 > wg.Wait() service.go
562 >
563 > if s.metricsHandler != nil {
564 > s.metricsHandler.Stop(s.logger)
565 > }
566
567 > s.logger.Info("frontend stopped") service.go
568 }
go.temporal.io/server/service/history/service.go 37 introduced LOC · 8 ranges

Open complete file

115
116 // Stop stops the service
117 > func (s *Service) Stop() { service.go
118 > s.readinessCancel()
119 >
120 > // remove self from membership ring and wait for traffic to drain
121 > var err error
122 > var waitTime time.Duration
123 > if align := s.config.AlignMembershipChange(); align > 0 {
124 propagation := s.membershipMonitor.ApproximateMaxPropagationTime()
125 asOf := util.NextAlignedTime(time.Now().Add(propagation), align)
126 s.logger.Info("ShutdownHandler: Evicting self from membership ring as of", tag.Timestamp(asOf))
127 waitTime, err = s.membershipMonitor.EvictSelfAt(asOf)
128 > } else { service.go
129 > s.logger.Info("ShutdownHandler: Evicting self from membership ring immediately")
130 > err = s.membershipMonitor.EvictSelf()
131 > }
132 > if err != nil {
133 s.logger.Error("ShutdownHandler: Failed to evict self from membership ring", tag.Error(err))
134 }
135 > s.healthServer.SetServingStatus(serviceName, healthpb.HealthCheckResponse_NOT_SERVING) service.go
136 >
137 > s.logger.Info("ShutdownHandler: Waiting for drain")
138 > if waitTime > 0 {
139 time.Sleep(
140 waitTime + // wait for membership change
142 s.config.ShardFinalizerTimeout(), // and then take this long to run a finalizer
143 )
144 > } else { service.go
145 > time.Sleep(s.config.ShutdownDrainDuration())
146 > }
147
148 > enableCloseInboundReplicationStreamOnShutdown := s.config.EnableCloseInboundReplicationStreamOnShutdown() service.go
149 > // When enabled, stop handler components (including the replication stream monitor) before
150 > // waiting for gRPC handlers to return. This signals inbound stream senders on the peer to
151 > // stop, allowing their handler goroutines to unblock and return cleanly before GracefulStop.
152 > // Without this, those goroutines block indefinitely and the gRPC server falls back to a
153 > // forceful Stop(), causing unclean H2 teardowns on the peer.
154 > // Guarded by feature flag so the ordering change can be reverted if needed.
155 > if enableCloseInboundReplicationStreamOnShutdown {
156 > s.logger.Info("ShutdownHandler: Initiating handler shutdown")
157 > s.handler.Stop()
158 > } else {
159 s.logger.Info("ShutdownHandler: Initiating shardController shutdown")
160 s.handler.controller.Stop()
162
163 // All grpc handlers should be cancelled now. Give them a little time to return.
164 > t := time.AfterFunc(2*time.Second, func() { service.go
165 s.logger.Info("ShutdownHandler: Drain time expired, stopping all traffic")
166 s.server.Stop()
167 })
168 > s.server.GracefulStop() service.go
169 > t.Stop()
170 > if !enableCloseInboundReplicationStreamOnShutdown {
171 s.handler.Stop()
172 }
173 > s.visibilityManager.Close() service.go
174 >
175 > s.logger.Info("history stopped")
176 }
go.temporal.io/server/service/matching/service.go 29 introduced LOC · 4 ranges

Open complete file

85
86 // Stop stops the service
87 > func (s *Service) Stop() { service.go
88 > // remove self from membership ring and wait for traffic to drain
89 > var err error
90 > var waitTime time.Duration
91 > if align := s.config.AlignMembershipChange(); align > 0 {
92 propagation := s.membershipMonitor.ApproximateMaxPropagationTime()
93 asOf := util.NextAlignedTime(time.Now().Add(propagation), align)
94 s.logger.Info("ShutdownHandler: Evicting self from membership ring as of", tag.Timestamp(asOf))
95 waitTime, err = s.membershipMonitor.EvictSelfAt(asOf)
96 > } else { service.go
97 > s.logger.Info("ShutdownHandler: Evicting self from membership ring immediately")
98 > err = s.membershipMonitor.EvictSelf()
99 > }
100 > if err != nil {
101 s.logger.Error("ShutdownHandler: Failed to evict self from membership ring", tag.Error(err))
102 }
103 > s.healthServer.SetServingStatus(serviceName, healthpb.HealthCheckResponse_NOT_SERVING) service.go
104 >
105 > s.logger.Info("ShutdownHandler: Waiting for others to discover I am unhealthy")
106 > time.Sleep(max(s.config.ShutdownDrainDuration(), waitTime))
107 >
108 > // At this point we should not get any new rpcs since we removed ourself from the ring.
109 > // Additionally, the engine will notice the membership change and stop all task queues
110 > // after a delay. However, we can do it immediately by stopping the handler (which stops
111 > // the engine which stops all task queues).
112 > s.handler.Stop()
113 >
114 > // All grpc handlers should be cancelled now. Give them a little time to return.
115 > t := time.AfterFunc(2*time.Second, func() {
116 s.logger.Info("ShutdownHandler: Drain time expired, stopping all traffic")
117 s.server.Stop()
118 })
119 > s.server.GracefulStop() service.go
120 > t.Stop()
121 >
122 > s.visibilityManager.Close()
123 >
124 > s.logger.Info("matching stopped")
125 }
go.temporal.io/server/temporal/fx.go 22 introduced LOC · 4 ranges

Open complete file

1105 func shutdownAll(exporters []otelsdktrace.SpanExporter) func(ctx context.Context) error {
1106 return func(ctx context.Context) error {
1107 > shutdownCtx, cancel := context.WithTimeout(context.Background(), 1*time.Second) fx.go
1108 > defer cancel()
1109 >
1110 > for _, e := range exporters {
1111 err := e.Shutdown(shutdownCtx)
1112 if errors.Is(err, context.DeadlineExceeded) {
1152 )
1153 }
1154 > case *fxevent.OnStopExecuting: fx.go
1155 > l.logger.Debug("OnStop hook executing",
1156 > tag.ComponentFX,
1157 > tag.String("callee", e.FunctionName),
1158 > tag.String("caller", e.CallerName),
1159 > )
1160 > case *fxevent.OnStopExecuted:
1161 > if e.Err != nil {
1162 l.logger.Error("OnStop hook failed",
1163 tag.ComponentFX,
1166 tag.Error(e.Err),
1167 )
1168 > } else { fx.go
1169 > l.logger.Debug("OnStop hook executed",
1170 > tag.ComponentFX,
1171 > tag.String("callee", e.FunctionName),
1172 > tag.String("caller", e.CallerName),
1173 > tag.Stringer("runtime", e.Runtime),
1174 > )
1175 > }
1176 case *fxevent.Supplied:
1177 if e.Err != nil {
1234 tag.ComponentFX,
1235 tag.Stringer("signal", e.Signal))
1236 > case *fxevent.Stopped: fx.go
1237 > if e.Err != nil {
1238 l.logger.Error("stop failed", tag.ComponentFX, tag.Error(e.Err))
1239 }
go.temporal.io/server/common/persistence/persistence_rate_limited_clients.go 21 introduced LOC · 7 ranges

Open complete file

262 }
263
264 > func (p *shardRateLimitedPersistenceClient) Close() { persistence_rate_limited_clients.go
265 > p.persistence.Close()
266 > }
267
268 func (p *executionRateLimitedPersistenceClient) GetName() string {
501 }
502
503 > func (p *executionRateLimitedPersistenceClient) Close() { persistence_rate_limited_clients.go
504 > p.persistence.Close()
505 > }
506
507 func (p *taskRateLimitedPersistenceClient) GetName() string {
637 }
638
639 > func (p *taskRateLimitedPersistenceClient) Close() { persistence_rate_limited_clients.go
640 > p.persistence.Close()
641 > }
642
643 func (p *metadataRateLimitedPersistenceClient) GetName() string {
755 }
756
757 > func (p *metadataRateLimitedPersistenceClient) Close() { persistence_rate_limited_clients.go
758 > p.persistence.Close()
759 > }
760
761 // AppendHistoryNodes add a node to history node table
998 }
999
1000 > func (p *queueRateLimitedPersistenceClient) Close() { persistence_rate_limited_clients.go
1001 > p.persistence.Close()
1002 > }
1003
1004 func (p *queueRateLimitedPersistenceClient) Init(
1009 }
1010
1011 > func (c *clusterMetadataRateLimitedPersistenceClient) Close() { persistence_rate_limited_clients.go
1012 > c.persistence.Close()
1013 > }
1014
1015 func (c *clusterMetadataRateLimitedPersistenceClient) GetName() string {
1100 }
1101
1102 > func (p *nexusEndpointRateLimitedPersistenceClient) Close() { persistence_rate_limited_clients.go
1103 > p.persistence.Close()
1104 > }
1105
1106 func (p *nexusEndpointRateLimitedPersistenceClient) GetNexusEndpoint(
go.temporal.io/server/service/history/replication/task_processor_manager.go 17 introduced LOC · 6 ranges

Open complete file

114 }
115
116 > func (r *taskProcessorManagerImpl) Stop() { task_processor_manager.go
117 > if !atomic.CompareAndSwapInt32(
118 > &r.status,
119 > common.DaemonStatusStarted,
120 > common.DaemonStatusStopped,
121 > ) {
122 return
123 }
124
125 > close(r.shutdownChan) task_processor_manager.go
126 >
127 > if r.enableFetcher {
128 r.shard.GetClusterMetadata().UnRegisterMetadataChangeCallback(r)
129 }
130 > r.taskProcessorLock.Lock() task_processor_manager.go
131 > for _, taskProcessors := range r.taskProcessors {
132 for _, processor := range taskProcessors {
133 processor.Stop()
134 }
135 }
136 > r.taskProcessorLock.Unlock() task_processor_manager.go
137 }
138
229 r.config.ReplicationTaskProcessorCleanupJitterCoefficient(shardID),
230 ))
231 > case <-r.shutdownChan: task_processor_manager.go
232 > return
233 }
234 }
243 r.checkReplicationDLQSize()
244 }
245 > case <-r.shutdownChan: task_processor_manager.go
246 > timer.Stop()
247 > return
248 }
249 }
go.temporal.io/server/service/worker/service.go 16 introduced LOC · 1 range

Open complete file

292
293 // Stop is called to stop the service
294 > func (s *Service) Stop() { service.go
295 > s.healthServer.SetServingStatus(ServiceName, healthpb.HealthCheckResponse_NOT_SERVING)
296 >
297 > s.scanner.Stop()
298 > s.perNamespaceWorkerManager.Stop()
299 > s.workerManager.Stop()
300 > s.visibilityManager.Close()
301 >
302 > s.server.GracefulStop()
303 >
304 > s.logger.Info(
305 > "worker service stopped",
306 > tag.ComponentWorker,
307 > tag.Address(s.hostInfo.GetAddress()),
308 > )
309 > }
310
311 func (s *Service) startParentClosePolicyProcessor() {
go.temporal.io/server/client/client_bean.go 13 introduced LOC · 1 range

Open complete file

120 // Close releases the resources held by the bean's clients. See the Bean
121 // interface for details. It is safe to call more than once.
122 > func (h *clientBeanImpl) Close() { client_bean.go
123 > h.clusterMetadata.UnRegisterMetadataChangeCallback(h)
124 >
125 > // The history and matching client wrapper chains implement Stop();
126 > // stopping them releases their daemon goroutines and cached gRPC
127 > // connections.
128 > if s, ok := h.historyClient.(interface{ Stop() }); ok {
129 > s.Stop()
130 > }
131 > if mc := h.matchingClient.Load(); mc != nil {
132 > if s, ok := mc.(interface{ Stop() }); ok {
133 > s.Stop()
134 > }
135 }
136 }
go.temporal.io/server/service/history/queues/scheduler.go 13 introduced LOC · 1 range

Open complete file

197 }
198
199 > func (s *schedulerImpl) Stop() { scheduler.go
200 > if s.channelWeightUpdateCh != nil {
201 > s.namespaceRegistry.UnregisterStateChangeCallback(s)
202 >
203 > // note we can't close the channelWeightUpdateCh here
204 > // as callback may still be triggered even after unregister returns
205 > // due to race condition
206 > //
207 > // channelWeightFn is only not nil when using host level scheduler
208 > // so Stop is only called when host is shutting down, and we don't need
209 > // to worry about open channels
210 > }
211 > s.Scheduler.Stop()
212 }
213
go.temporal.io/server/service/history/queues/queue_scheduled.go 12 introduced LOC · 4 ranges

Open complete file

139 }
140
141 > func (p *scheduledQueue) Stop() { queue_scheduled.go
142 > if !atomic.CompareAndSwapInt32(&p.status, common.DaemonStatusStarted, common.DaemonStatusStopped) {
143 return
144 }
145
146 > p.logger.Info("", tag.LifeCycleStopping) queue_scheduled.go
147 > defer p.logger.Info("", tag.LifeCycleStopped)
148 >
149 > close(p.shutdownCh)
150 > p.timerGate.Close()
151 >
152 > if success := common.AwaitWaitGroup(&p.shutdownWG, time.Minute); !success {
153 p.logger.Warn("", tag.LifeCycleStopTimedout)
154 }
155
156 > p.queueBase.Stop() queue_scheduled.go
157 }
158
184
185 select {
186 > case <-p.shutdownCh: queue_scheduled.go
187 > return
188 case <-p.newTimerCh:
189 metrics.NewTimerNotifyCounter.With(p.metricsHandler).Record(1)
go.temporal.io/server/service/history/handler.go 11 introduced LOC · 2 ranges

Open complete file

191
192 // Stop stops the handler
193 > func (h *Handler) Stop() { handler.go
194 > if !atomic.CompareAndSwapInt32(
195 > &h.status,
196 > common.DaemonStatusStarted,
197 > common.DaemonStatusStopped,
198 > ) {
199 return
200 }
201
202 > h.streamReceiverMonitor.Stop() handler.go
203 > h.replicationTaskFetcherFactory.Stop()
204 > h.controller.Stop()
205 > h.eventNotifier.Stop()
206 > h.dlqMetricsEmitter.Stop()
207 }
208
go.temporal.io/server/service/history/replication/task_fetcher.go 11 introduced LOC · 3 ranges

Open complete file

119
120 // Stop stops the fetchers
121 > func (f *taskFetcherFactoryImpl) Stop() { task_fetcher.go
122 > if !atomic.CompareAndSwapInt32(
123 > &f.status,
124 > common.DaemonStatusStarted,
125 > common.DaemonStatusStopped,
126 > ) {
127 return
128 }
129
130 > f.clusterMetadata.UnRegisterMetadataChangeCallback(f) task_fetcher.go
131 > f.fetchersLock.Lock()
132 > defer f.fetchersLock.Unlock()
133 > for _, fetcher := range f.fetchers {
134 fetcher.Stop()
135 }
136 > f.logger.Info("Replication task fetchers stopped.") task_fetcher.go
137 }
138
go.temporal.io/server/common/deadlock/deadlock.go 10 introduced LOC · 2 ranges

Open complete file

100 }
101
102 > func (dd *deadlockDetector) Stop() error { deadlock.go
103 > for _, pool := range dd.pools {
104 > pool.Stop()
105 > }
106 > dd.loops.Cancel()
107 > // don't wait for workers to exit, they may be blocked
108 > return nil
109 }
110
160 select {
161 case <-timer.C:
162 > case <-ctx.Done(): deadlock.go
163 > timer.Stop()
164 > return ctx.Err()
165 }
166 }
go.temporal.io/server/service/frontend/workflow_handler.go 10 introduced LOC · 1 range

Open complete file

429
430 // Stop stops the handler
431 > func (wh *WorkflowHandler) Stop() { workflow_handler.go
432 > if atomic.CompareAndSwapInt32(
433 > &wh.status,
434 > common.DaemonStatusStarted,
435 > common.DaemonStatusStopped,
436 > ) {
437 > wh.namespaceRegistry.UnregisterStateChangeCallback(wh)
438 > wh.healthServer.SetServingStatus(WorkflowServiceName, healthpb.HealthCheckResponse_NOT_SERVING)
439 > wh.healthInterceptor.SetHealthy(false)
440 > }
441 }
442
go.temporal.io/server/service/worker/worker.go 10 introduced LOC · 2 ranges

Open complete file

101 }
102
103 > func (wm *workerManager) Stop() { worker.go
104 > if !atomic.CompareAndSwapInt32(
105 > &wm.status,
106 > common.DaemonStatusStarted,
107 > common.DaemonStatusStopped,
108 > ) {
109 return
110 }
111
112 > for _, w := range wm.workers { worker.go
113 > w.Stop()
114 > }
115 > wm.logger.Info("", tag.ComponentWorkerManager, tag.LifeCycleStopped)
116 }
go.temporal.io/server/service/history/history_engine.go 9 introduced LOC · 1 range

Open complete file

373 }
374
375 > e.logger.Info("", tag.LifeCycleStopping) history_engine.go
376 > defer e.logger.Info("", tag.LifeCycleStopped)
377 >
378 > for _, queueProcessor := range e.queueProcessors {
379 > queueProcessor.Stop()
380 > }
381 > e.replicationProcessorMgr.Stop()
382 > // unset the failover callback
383 > e.shardContext.GetNamespaceRegistry().UnregisterStateChangeCallback(e)
384 }
385
go.temporal.io/server/service/history/queue_factory_base.go 9 introduced LOC · 2 ranges

Open complete file

193 return nil
194 },
195 > OnStop: func(context.Context) error { queue_factory_base.go
196 > for _, factory := range params.Factories {
197 > factory.Stop()
198 > }
199 > return nil
200 },
201 },
209 }
210
211 > func (f *QueueFactoryBase) Stop() { queue_factory_base.go
212 > if f.HostScheduler != nil {
213 > f.HostScheduler.Stop()
214 > }
215 }
216
go.temporal.io/server/client/history/connections.go 8 introduced LOC · 3 ranges

Open complete file

74
75 // Close stops the watcher and closes all pooled connections.
76 > func (c *connectionPoolImpl[C]) Close() { connections.go
77 > if !c.closed.CompareAndSwap(false, true) {
78 return
79 }
80 > c.watcher.Cancel() connections.go
81 > <-c.watcher.Done()
82 > // Set closed before reaping so a concurrent create can't re-cache a conn.
83 > c.conns.Range(func(key, value any) bool {
84 c.conns.Delete(key)
85 if err := value.(clientConnection[C]).grpcConn.Close(); err != nil {
106 for {
107 select {
108 > case <-ctx.Done(): connections.go
109 > return nil
110 case event := <-ch:
111 for _, h := range event.HostsRemoved {
go.temporal.io/server/client/matching/client.go 8 introduced LOC · 2 ranges

Open complete file

87 // the eviction watcher and partition-cache rotation goroutines and closes every
88 // cached gRPC connection. It is safe to call more than once.
89 > func (c *clientImpl) Stop() { client.go
90 > c.evictionWatcher.Cancel()
91 > <-c.evictionWatcher.Done()
92 > c.partitionCache.Stop()
93 > c.clients.EvictAll()
94 > }
95
96 // watchMembership evicts cached clients whose host leaves the membership ring.
112 for {
113 select {
114 > case <-ctx.Done(): client.go
115 > return nil
116 case event := <-ch:
117 for _, h := range event.HostsRemoved {
go.temporal.io/server/common/client_cache.go 7 introduced LOC · 1 range

Open complete file

131 }
132
133 > func (c *clientCacheImpl) EvictAll() { client_cache.go
134 > c.cacheLock.Lock()
135 > entries := c.clients
136 > c.clients = make(map[string]cachedEntry)
137 > c.cacheLock.Unlock()
138 >
139 > for _, entry := range entries {
140 if entry.release != nil {
141 if err := entry.release(); err != nil {

Direct hierarchy

Parents

Children