122
123
// ScavengeBuildIds scans all task queue user data entries in all namespaces and cleans up unused build ids.
124
>
func (a *Activities) ScavengeBuildIds(ctx context.Context, input BuildIdScavangerInput) error {
scavenger.go
125
>
a.setDefaults(&input)
126
>
127
>
var heartbeat heartbeatDetails
128
>
if activity.HasHeartbeatDetails(ctx) {
129
>
if err := activity.GetHeartbeatDetails(ctx, &heartbeat); err != nil {
130
return temporal.NewNonRetryableApplicationError("failed to load previous heartbeat details", "TypeError", err)
131
}
132
}
133
>
rateLimiter := quotas.NewDefaultOutgoingRateLimiter(quotas.RateFn(a.buildIdScavengerVisibilityRPS))
scavenger.go
134
>
for {
135
>
nsResponse, err := a.metadataManager.ListNamespaces(ctx, &persistence.ListNamespacesRequest{
136
>
PageSize: input.NamespaceListPageSize,
137
>
NextPageToken: heartbeat.NamespaceNextPageToken,
138
>
IncludeDeleted: false, // Don't care about deleted namespaces.
139
>
})
140
>
if err != nil {
141
return err
142
}
143
>
for heartbeat.NamespaceIdx < len(nsResponse.Namespaces) {
scavenger.go
144
>
nsId := nsResponse.Namespaces[heartbeat.NamespaceIdx].Namespace.Info.Id
145
>
if err := a.processNamespaceEntry(ctx, rateLimiter, input, &heartbeat, nsId); err != nil {
146
return err
147
}
149
>
a.recordHeartbeat(ctx, heartbeat)
150
}
152
>
heartbeat.NamespaceNextPageToken = nsResponse.NextPageToken
153
>
if len(heartbeat.NamespaceNextPageToken) == 0 {
154
>
break
155
}
157
}
159
}
160