ronantakizawa/codereview-bench huggingface dataset:
Repository: hashicorp/nomad
File: client/fingerprint/storage.go (Go)

Code before:
}

func NewStorageFingerprint(logger log.Logger) Fingerprint {
	fp := &StorageFingerprint{logger: logger.Named("storage")}
	return fp
}

func (f *StorageFingerprint) Fingerprint(req *FingerprintRequest, resp *FingerprintResponse) error {
	cfg := req.Config

	// Guard against unset AllocDir
	storageDir := cfg.AllocDir
	if storageDir == "" {
		var err error
		storageDir, err = os.Getwd()
		if err != nil {
			return fmt.Errorf("unable to get CWD from filesystem: %s", err)
		}
	}

	volume, total, err := f.diskInfo(storageDir)
	if err != nil {
		return fmt.Errorf("failed to determine disk space for %s: %v", storageDir, err)
	}

	if cfg.DiskTotalMB > 0 {
		total = uint64(cfg.DiskTotalMB) * bytesPerMegabyte
	}

	free := total - f.reservedDisk(req)

	// DEPRECATED: remove in 1.13.0
	if cfg.DiskFreeMB > 0 {
		f.logger.Warn("disk_free_mb is deprecated, please use reserved.disk")
		free = uint64(cfg.DiskFreeMB) * bytesPerMegabyte
	}

	resp.AddAttribute("unique.storage.volume", volume)
	resp.AddAttribute("unique.storage.bytestotal", strconv.FormatUint(total, 10))
	resp.AddAttribute("unique.storage.bytesfree", strconv.FormatUint(free, 10))

	// set the disk size for the response
	resp.NodeResources = &structs.NodeResources{
		Disk: structs.NodeDiskResources{
			DiskMB: int64(total / bytesPerMegabyte),
		},
	}
	resp.Detected = true

	return nil
}

Review comment (bug):
I wonder if we should also deprecate this field as well. It was added in the same PR as `DiskFreeMB`. I think the only 2 use cases for this would be if we could not detect the clients total disk space for some reason (but I think this would be a Nomad bug) or if a user wanted to "overprovision" their storage for scheduling sake.

Code after:
}

func NewStorageFingerprint(logger log.Logger) Fingerprint {
	fp := &StorageFingerprint{logger: logger.Named("storage")}
	return fp
}

func (f *StorageFingerprint) Fingerprint(req *FingerprintRequest, resp *FingerprintResponse) error {
	cfg := req.Config

	// Guard against unset AllocDir
	storageDir := cfg.AllocDir
	if storageDir == "" {
		var err error
		storageDir, err = os.Getwd()
		if err != nil {
			return fmt.Errorf("unable to get CWD from filesystem: %s", err)
		}
	}

	volume, total, err := f.diskInfo(storageDir)
	if err != nil {
		return fmt.Errorf("failed to determine disk space for %s: %v", storageDir, err)
	}

	if cfg.DiskTotalMB > 0 {
		total = uint64(cfg.DiskTotalMB) * bytesPerMegabyte
	}

	resp.AddAttribute("unique.storage.volume", volume)
	resp.AddAttribute("unique.storage.bytestotal", strconv.FormatUint(total, 10))

	// set the disk size for the response
	resp.NodeResources = &structs.NodeResources{
		Disk: structs.NodeDiskResources{
			DiskMB: int64(total / bytesPerMegabyte),
		},
	}
	resp.Detected = true

	return nil
}
Repository: hashicorp/nomad
File: command/agent/agent_endpoint.go (Go)

Code before:
		NumSchedulers:     args.NumSchedulers,
		EnabledSchedulers: args.EnabledSchedulers,
	}
	if newArgs.IsInvalid() {
		return nil, CodedError(http.StatusBadRequest, "Invalid request")
	}
	reply := srv.SetSchedulerWorkerConfig(newArgs)

	response := &api.AgentSchedulerWorkerConfigResponse{
		ServerID:          srv.LocalMember().Name,
		NumSchedulers:     reply.NumSchedulers,
		EnabledSchedulers: reply.EnabledSchedulers,
	}

	return response, nil
}

func checkPermissions(srv *nomad.Server, secret string, perm func(acl *acl.ACL) bool) error {
	r := &structs.GenericRequest{}
	r.AuthToken = secret
	if authErr := srv.Authenticate(nil, r); authErr != nil {
		return CodedError(http.StatusInternalServerError, authErr.Error())
	}

	aclObj, err := srv.ResolveACL(r)
	if err != nil {
		return CodedError(http.StatusInternalServerError, err.Error())
	} else if !perm(aclObj) {
		return CodedError(http.StatusForbidden, structs.ErrPermissionDenied.Error())
	}

	return nil
}


Review comment (security):
In server RPC endpoints these errors are always returned as `ErrPermissionDenied`. We should use the 403 code here.

Code after:
		NumSchedulers:     args.NumSchedulers,
		EnabledSchedulers: args.EnabledSchedulers,
	}
	if newArgs.IsInvalid() {
		return nil, CodedError(http.StatusBadRequest, "Invalid request")
	}
	reply := srv.SetSchedulerWorkerConfig(newArgs)

	response := &api.AgentSchedulerWorkerConfigResponse{
		ServerID:          srv.LocalMember().Name,
		NumSchedulers:     reply.NumSchedulers,
		EnabledSchedulers: reply.EnabledSchedulers,
	}

	return response, nil
}

// aclPermissionCheckHelper takes a token string and checks it with Authenticate
// and ResolveACL methods. If the token doesn't satisfy perm function, an error
// is returned.
func aclPermissionCheckHelper(srv *nomad.Server, secret string, perm func(acl *acl.ACL) bool) error {
	r := &structs.GenericRequest{}
	r.AuthToken = secret
	if authErr := srv.Authenticate(nil, r); authErr != nil {
		return CodedError(http.StatusForbidden, authErr.Error())
	}

	aclObj, err := srv.ResolveACL(r)
	if err != nil {
		return CodedError(http.StatusInternalServerError, err.Error())
	} else if !perm(aclObj) {
		return CodedError(http.StatusForbidden, structs.ErrPermissionDenied.Error())
	}

	return nil
}
Repository: hashicorp/nomad
File: api/event_stream_test.go (Go)

Code before:

	streamCh, err := events.Stream(ctx, topics, 0, q)
	must.NoError(t, err)

	select {
	case event := <-streamCh:
		if event.Err != nil {
			must.Unreachable(t, must.Sprintf("unexpected %v", event.Err))
		}
		must.Len(t, 1, event.Events)
		must.Eq(t, "Evaluation", string(event.Events[0].Topic))
	case <-time.After(5 * time.Second):
		must.Unreachable(t, must.Sprint("failed waiting for event stream event"))
	}

	// Stop the server to ensure EOF is returned
	s.Stop()

	for {
		select {
		case event, ok := <-streamCh:
			if !ok {
				must.Unreachable(t, must.Sprintf("chan closed before EOF received"))
			}

			// Sadly decode doesn't return io.EOF
			if event.Err != nil && strings.HasSuffix(event.Err.Error(), "EOF") {
				// Succcess! Make sure chan gets closed
				select {
				case _, ok := <-streamCh:
					if ok {
						must.Unreachable(t, must.Sprintf("expected chan to close after EOF"))
					}

					// Success!
					return
				case <-time.After(5 * time.Second):
					must.Unreachable(t, must.Sprint("failed waiting for event stream to close"))
				}
			}

			if event.Err != nil {
				must.Unreachable(t, must.Sprintf("unexpected %v (%T)", event.Err, event.Err))
			}
			must.Len(t, 1, event.Events)
			must.Eq(t, "Evaluation", string(event.Events[0].Topic))
		case <-time.After(5 * time.Second):
			must.Unreachable(t, must.Sprint("failed waiting for event stream EOF"))
		}
	}
}

Review comment (suggestion):
Poking through this, it looks like decode is kicking back an `io.ErrUnexpectedEOF` and swapping this out with an `errors.Is(event.Err, io.ErrUnexpectedEOF)` works as expected.

Code after:
	defer cancel()

	streamCh, err := events.Stream(ctx, topics, 0, q)
	must.NoError(t, err)

	select {
	case event := <-streamCh:
		if event.Err != nil {
			must.Unreachable(t, must.Sprintf("unexpected %v", event.Err))
		}
		must.Len(t, 1, event.Events)
		must.Eq(t, "Evaluation", string(event.Events[0].Topic))
	case <-time.After(5 * time.Second):
		must.Unreachable(t, must.Sprint("failed waiting for event stream event"))
	}

	// Stop the server to ensure EOF is returned
	s.Stop()

	for {
		select {
		case event, ok := <-streamCh:
			if !ok {
				must.Unreachable(t, must.Sprintf("chan closed before EOF received"))
			}

			if event.Err != nil && errors.Is(event.Err, io.ErrUnexpectedEOF) {
				// Succcess! Make sure chan gets closed
				select {
				case _, ok := <-streamCh:
					if ok {
						must.Unreachable(t, must.Sprintf("expected chan to close after EOF"))
					}

					// Success!
					return
				case <-time.After(5 * time.Second):
					must.Unreachable(t, must.Sprint("failed waiting for event stream to close"))
				}
			}

			if event.Err != nil {
				must.Unreachable(t, must.Sprintf("unexpected %v (%T)", event.Err, event.Err))
			}
			must.Len(t, 1, event.Events)
			must.Eq(t, "Evaluation", string(event.Events[0].Topic))
		case <-time.After(5 * time.Second):
			must.Unreachable(t, must.Sprint("failed waiting for event stream EOF"))
		}
	}
}
Repository: hashicorp/nomad
File: drivers/qemu/driver.go (Go)

Code before:
		}
	}

	return false
}

// hardcoded list of drive interfaces, Qemu currently supports
var allowedDriveInterfaces = []string{"ide", "scsi", "sd", "mtd", "floppy", "pflash", "virtio", "none"}

func isAllowedDriveInterface(driveInterface string) bool {
	for _, ai := range allowedDriveInterfaces {
		if driveInterface == ai {
			return true
		}
	}

	return false
}

func validateEmulator(emulator string, allowedEmulators []string) error {
	if len(allowedEmulators) > 0 {
		if !slices.Contains(allowedEmulators, emulator) {
			return fmt.Errorf("emulator '%s' is not an allowed emulator", emulator)
		}
	} else {
		match, err := regexp.MatchString("qemu-system-*", emulator)
		if err != nil {
			return err
		}
		if !match {
			return fmt.Errorf("emulator '%s' is not valid", emulator)
		}
	}
	return nil
}

// validateArgs ensures that all QEMU command line params are in the
// allowlist. This function must be called after all interpolation has
// taken place.
func validateArgs(pluginConfigAllowList, args []string) error {
	if len(pluginConfigAllowList) > 0 {
		allowed := map[string]struct{}{}
		for _, arg := range pluginConfigAllowList {
			allowed[arg] = struct{}{}
		}
		for _, arg := range args {
			if strings.HasPrefix(strings.TrimSpace(arg), "-") {
				if _, ok := allowed[arg]; !ok {
					return fmt.Errorf("%q is not in args_allowlist", arg)
				}
			}

Review comment (suggestion):
If we're not using  the simpler `github.com/ryanuber/go-glob` here, we might want to use [path.Match](https://pkg.go.dev/path#Match) instead of `regexp`, which would prevent inclusion of `/` in the path.

Code after:
		}
	}

	for _, f := range bins {
		em := strings.TrimPrefix(filepath.Base(f), "qemu-system-")
		if err := validateEmulator(em, allowList); err != nil {
			continue
		}
		emulators = append(emulators, em)
	}

	return emulators
}

func isAllowedImagePath(allowedPaths []string, allocDir, imagePath string) bool {
	if !filepath.IsAbs(imagePath) {
		imagePath = filepath.Join(allocDir, imagePath)
	}

	isParent := func(parent, path string) bool {
		rel, err := filepath.Rel(parent, path)
		return err == nil && !strings.HasPrefix(rel, "..")
	}

	// check if path is under alloc dir
	if isParent(allocDir, imagePath) {
		return true
	}

	// check allowed paths
	for _, ap := range allowedPaths {
		if isParent(ap, imagePath) {
			return true
		}
	}

	return false
}

// hardcoded list of drive interfaces, Qemu currently supports
var allowedDriveInterfaces = []string{"ide", "scsi", "sd", "mtd", "floppy", "pflash", "virtio", "none"}

func isAllowedDriveInterface(driveInterface string) bool {
	return slices.Contains(allowedDriveInterfaces, driveInterface)
}

// validateEmulator validate whether the specified emulator is in allowedEmulators
func validateEmulator(emulator string, allowedEmulators []string) error {
	if len(allowedEmulators) > 0 {
		if !slices.Contains(allowedEmulators, emulator) {
			return fmt.Errorf("'%s' is not an allowed emulator", emulator)
Repository: hashicorp/nomad
File: nomad/state/events.go (Go)

Code before:
				ACLBindingRule: after,
			},
		}, true
	case "evals":
		after, ok := change.After.(*structs.Evaluation)
		if !ok {
			return structs.Event{}, false
		}
		return structs.Event{
			Topic: structs.TopicEvaluation,
			Key:   after.ID,
			FilterKeys: []string{
				after.JobID,
				after.DeploymentID,
			},
			Namespace: after.Namespace,
			Payload: &structs.EvaluationEvent{
				Evaluation: after,
			},
		}, true
	case "allocs":
		after, ok := change.After.(*structs.Allocation)
		if !ok {
			return structs.Event{}, false
		}
		alloc := after.CopySkipJobAndSanitize()

		filterKeys := []string{
			alloc.JobID,
			alloc.DeploymentID,
		}

		return structs.Event{
			Topic:      structs.TopicAllocation,
			Key:        after.ID,
			FilterKeys: filterKeys,
			Namespace:  after.Namespace,
			Payload:    &structs.AllocationEvent{Allocation: alloc},
		}, true
	case "jobs":
		after, ok := change.After.(*structs.Job)
		if !ok {
			return structs.Event{}, false
		}
		return structs.Event{
			Topic:     structs.TopicJob,
			Key:       after.ID,
			Namespace: after.Namespace,
			Payload: &structs.JobEvent{
				Job: after,
			},

Review comment (security):
Rather than introducing a new method and the flags on the existing methods, could we just move the call to `Sanitize()` here? That's what the node events do.

Code after:
				ACLBindingRule: after,
			},
		}, true
	case "evals":
		after, ok := change.After.(*structs.Evaluation)
		if !ok {
			return structs.Event{}, false
		}
		return structs.Event{
			Topic: structs.TopicEvaluation,
			Key:   after.ID,
			FilterKeys: []string{
				after.JobID,
				after.DeploymentID,
			},
			Namespace: after.Namespace,
			Payload: &structs.EvaluationEvent{
				Evaluation: after,
			},
		}, true
	case "allocs":
		after, ok := change.After.(*structs.Allocation)
		if !ok {
			return structs.Event{}, false
		}
		alloc := after.Sanitize()

		filterKeys := []string{
			alloc.JobID,
			alloc.DeploymentID,
		}

		// remove job info to help keep size of alloc event down
		alloc.Job = nil

		return structs.Event{
			Topic:      structs.TopicAllocation,
			Key:        after.ID,
			FilterKeys: filterKeys,
			Namespace:  after.Namespace,
			Payload:    &structs.AllocationEvent{Allocation: alloc},
		}, true
	case "jobs":
		after, ok := change.After.(*structs.Job)
		if !ok {
			return structs.Event{}, false
		}
		return structs.Event{
			Topic:     structs.TopicJob,
			Key:       after.ID,
			Namespace: after.Namespace,
Repository: hashicorp/nomad
File: command/agent/node_endpoint.go (Go)

Code before:
func (s *HTTPServer) nodeAllocations(resp http.ResponseWriter, req *http.Request,
	nodeID string) (interface{}, error) {
	if req.Method != http.MethodGet {
		return nil, CodedError(405, ErrInvalidMethod)
	}
	args := structs.NodeSpecificRequest{
		NodeID: nodeID,
	}
	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
		return nil, nil
	}

	var out structs.NodeAllocsResponse
	if err := s.agent.RPC("Node.GetAllocs", &args, &out); err != nil {
		return nil, err
	}

	setMeta(resp, &out.QueryMeta)
	if out.Allocs == nil {
		out.Allocs = make([]*structs.Allocation, 0)
	}
	for _, alloc := range out.Allocs {
		if alloc.SignedIdentities != nil {
			alloc = alloc.Sanitize()
		}
		alloc.SetEventDisplayMessages()
	}
	return out.Allocs, nil
}

func (s *HTTPServer) nodeToggleDrain(resp http.ResponseWriter, req *http.Request,
	nodeID string) (interface{}, error) {
	if req.Method != http.MethodPut && req.Method != http.MethodPost {
		return nil, CodedError(405, ErrInvalidMethod)
	}

	var drainRequest api.NodeUpdateDrainRequest

	err := decodeBody(req, &drainRequest)
	if err != nil {
		return nil, CodedError(400, err.Error())
	}

	args := structs.NodeUpdateDrainRequest{
		NodeID:       nodeID,
		MarkEligible: drainRequest.MarkEligible,
		Meta:         drainRequest.Meta,
	}
	if drainRequest.DrainSpec != nil {
		args.DrainStrategy = &structs.DrainStrategy{
			DrainSpec: structs.DrainSpec{

Review comment (bug):
It looks like this mutates the alloc too! I think this is relying on the alloc being copied by `Sanitize`, which is probably a bug given that we know it might not have been copied. So maybe we should remove the nil check here?

Code after:
func (s *HTTPServer) nodeAllocations(resp http.ResponseWriter, req *http.Request,
	nodeID string) (interface{}, error) {
	if req.Method != http.MethodGet {
		return nil, CodedError(405, ErrInvalidMethod)
	}
	args := structs.NodeSpecificRequest{
		NodeID: nodeID,
	}
	if s.parse(resp, req, &args.Region, &args.QueryOptions) {
		return nil, nil
	}

	var out structs.NodeAllocsResponse
	if err := s.agent.RPC("Node.GetAllocs", &args, &out); err != nil {
		return nil, err
	}

	setMeta(resp, &out.QueryMeta)
	if out.Allocs == nil {
		out.Allocs = make([]*structs.Allocation, 0)
	}

	sanitizedAllocs := []*structs.Allocation{}
	for _, alloc := range out.Allocs {
		alloc = alloc.Sanitize()
		alloc.SetEventDisplayMessages()
		sanitizedAllocs = append(sanitizedAllocs, alloc)
	}
	return sanitizedAllocs, nil
}

func (s *HTTPServer) nodeToggleDrain(resp http.ResponseWriter, req *http.Request,
	nodeID string) (interface{}, error) {
	if req.Method != http.MethodPut && req.Method != http.MethodPost {
		return nil, CodedError(405, ErrInvalidMethod)
	}

	var drainRequest api.NodeUpdateDrainRequest

	err := decodeBody(req, &drainRequest)
	if err != nil {
		return nil, CodedError(400, err.Error())
	}

	args := structs.NodeUpdateDrainRequest{
		NodeID:       nodeID,
		MarkEligible: drainRequest.MarkEligible,
		Meta:         drainRequest.Meta,
	}
	if drainRequest.DrainSpec != nil {
		args.DrainStrategy = &structs.DrainStrategy{
Repository: hashicorp/nomad
File: nomad/stream/event_broker.go (Go)

Code before:
	s.mu.RLock()
	defer s.mu.RUnlock()

	for _, secretID := range tokenSecretIDs {
		if subs, ok := s.byToken[secretID]; ok {
			for req, sub := range subs {
				// close and remove subscriptions from the map
				sub.forceClose()
				delete(subs, req)
			}

			// once all subscriptions are closed, remove the token from the map
			delete(s.byToken, secretID)
		}
	}
}

func (s *subscriptions) closeSubscriptionFunc(tokenSecretID string, fn func(*Subscription) bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// early exit
	subs, ok := s.byToken[tokenSecretID]
	if !ok {
		return
	}

	for req, sub := range s.byToken[tokenSecretID] {
		if fn(sub) {
			sub.forceClose()
			delete(subs, req)
		}
	}

	// If no subs remain for this token, delete the key.
	if len(subs) == 0 {
		delete(s.byToken, tokenSecretID)
	}
}

// unsubscribeFn returns a function that the subscription will call to remove
// itself from the subsByToken.
// This function is returned as a closure so that the caller doesn't need to keep
// track of the SubscriptionRequest, and can not accidentally call unsubscribeFn with the
// wrong pointer.
func (s *subscriptions) unsubscribeFn(req *SubscribeRequest) func() {
	return func() {
		s.mu.Lock()
		defer s.mu.Unlock()

		subsByToken, ok := s.byToken[req.Token]

Review comment (security):
We can use this `subs` in the `range` expression below, rather than re-hashing the `tokenSecretID`

Code after:
	s.mu.Lock()
	defer s.mu.Unlock()

	for _, secretID := range tokenSecretIDs {
		if subs, ok := s.byToken[secretID]; ok {
			for req, sub := range subs {
				// close and remove subscriptions from the map
				sub.forceClose()
				delete(subs, req)
			}

			// once all subscriptions are closed, remove the token from the map
			delete(s.byToken, secretID)
		}
	}
}

func (s *subscriptions) closeSubscriptionFunc(tokenSecretID string, fn func(*Subscription) bool) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// early exit
	subs, ok := s.byToken[tokenSecretID]
	if !ok {
		return
	}

	for req, sub := range subs {
		if fn(sub) {
			sub.forceClose()
			delete(subs, req)
		}
	}

	// If no subs remain for this token, delete the key.
	if len(subs) == 0 {
		delete(s.byToken, tokenSecretID)
	}
}

// unsubscribeFn returns a function that the subscription will call to remove
// itself from the subsByToken.
// This function is returned as a closure so that the caller doesn't need to keep
// track of the SubscriptionRequest, and can not accidentally call unsubscribeFn with the
// wrong pointer.
func (s *subscriptions) unsubscribeFn(req *SubscribeRequest) func() {
	return func() {
		s.mu.Lock()
		defer s.mu.Unlock()

		subsByToken, ok := s.byToken[req.Token]
Repository: hashicorp/nomad
File: nomad/stream/event_broker.go (Go)

Code before:
func (s *subscriptions) add(req *SubscribeRequest, sub *Subscription) {
	s.mu.Lock()
	defer s.mu.Unlock()

	subsByToken, ok := s.byToken[req.Token]
	if !ok {
		subsByToken = make(map[*SubscribeRequest]*Subscription)
		s.byToken[req.Token] = subsByToken
	}
	subsByToken[req] = sub
}

func (s *subscriptions) closeSubscriptionsForTokens(tokenSecretIDs []string) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	for _, secretID := range tokenSecretIDs {
		if subs, ok := s.byToken[secretID]; ok {
			for req, sub := range subs {
				// close and remove subscriptions from the map
				sub.forceClose()
				delete(subs, req)
			}

			// once all subscriptions are closed, remove the token from the map
			delete(s.byToken, secretID)
		}
	}
}

func (s *subscriptions) closeSubscriptionFunc(tokenSecretID string, fn func(*Subscription) bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// early exit
	subs, ok := s.byToken[tokenSecretID]
	if !ok {
		return
	}

	for req, sub := range s.byToken[tokenSecretID] {
		if fn(sub) {
			sub.forceClose()
			delete(subs, req)
		}
	}

	// If no subs remain for this token, delete the key.
	if len(subs) == 0 {
		delete(s.byToken, tokenSecretID)
	}

Review comment (suggestion):
We're only holding read lock on `s.mu` at this point. If we're going to delete here we need to upgrade that to a write lock.

Code after:
func (s *subscriptions) add(req *SubscribeRequest, sub *Subscription) {
	s.mu.Lock()
	defer s.mu.Unlock()

	subsByToken, ok := s.byToken[req.Token]
	if !ok {
		subsByToken = make(map[*SubscribeRequest]*Subscription)
		s.byToken[req.Token] = subsByToken
	}
	subsByToken[req] = sub
}

func (s *subscriptions) closeSubscriptionsForTokens(tokenSecretIDs []string) {
	s.mu.Lock()
	defer s.mu.Unlock()

	for _, secretID := range tokenSecretIDs {
		if subs, ok := s.byToken[secretID]; ok {
			for req, sub := range subs {
				// close and remove subscriptions from the map
				sub.forceClose()
				delete(subs, req)
			}

			// once all subscriptions are closed, remove the token from the map
			delete(s.byToken, secretID)
		}
	}
}

func (s *subscriptions) closeSubscriptionFunc(tokenSecretID string, fn func(*Subscription) bool) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// early exit
	subs, ok := s.byToken[tokenSecretID]
	if !ok {
		return
	}

	for req, sub := range subs {
		if fn(sub) {
			sub.forceClose()
			delete(subs, req)
		}
	}

	// If no subs remain for this token, delete the key.
	if len(subs) == 0 {
		delete(s.byToken, tokenSecretID)
	}