90
// Returns the count of added and removed entries separately.
91
// Workers with WORKER_STATUS_SHUTDOWN are immediately removed from the registry.
92
>
func (b *bucket) upsertHeartbeats(nsID namespace.ID, nsName namespace.Name, principal *commonpb.Principal, heartbeats []*workerpb.WorkerHeartbeat) (added int64, removed int64) {
registry_impl.go
93
>
now := time.Now()
94
>
95
>
b.mu.Lock()
96
>
defer b.mu.Unlock()
97
>
98
>
ns, ok := b.namespaces[nsID]
99
>
if !ok {
100
>
ns = &nsEntries{name: nsName, workers: make(map[string]*entry)}
101
>
b.namespaces[nsID] = ns
102
>
}
103
>
ns.name = nsName
104
>
105
>
for _, hb := range heartbeats {
106
>
key := hb.WorkerInstanceKey
107
>
108
>
// If worker is shutting down, remove it immediately
109
>
if hb.Status == enumspb.WORKER_STATUS_SHUTDOWN {
110
if e, exists := ns.workers[key]; exists {
111
b.order.Remove(e.elem)