ronantakizawa/codereview-bench huggingface dataset:
Repository: hashicorp/nomad
File: nomad/stream/event_broker.go (Go)

Code before:
			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]
		if !ok {
			return
		}

		sub := subsByToken[req]
		if sub == nil {
			return
		}

		// close the subscription
		sub.forceClose()


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 too.

Code after:
			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]
		if !ok {
			return
		}

		sub := subsByToken[req]
		if sub == nil {
			return
		}

		// close the subscription
		sub.forceClose()
Repository: hashicorp/nomad
File: drivers/qemu/driver.go (Go)

Code before:

	return ch, nil
}

func (d *Driver) StopTask(taskID string, timeout time.Duration, signal string) error {
	handle, ok := d.tasks.Get(taskID)
	if !ok {
		return drivers.ErrTaskNotFound
	}

	// Attempt a graceful shutdown only if it was configured in the job
	if handle.monitorPath != "" {
		err := sendQemuShutdown(d.logger, handle.monitorPath, handle.pid)
		if err == nil {
			ctx, cancel := context.WithTimeout(context.Background(), timeout)
			defer cancel()
			ticker := time.NewTicker(1 * time.Second)
			defer ticker.Stop()
		out:
			for {
				select {
				case <-ctx.Done():
					d.logger.Error("graceful shutdown", "pid", handle.pid, "timeout after", timeout)
					break out
				case <-ticker.C:
					handle, _ = d.tasks.Get(taskID)
					if handle.procState != drivers.TaskStateRunning {
						break out
					}
				}
			}
		} else {
			d.logger.Debug("error sending graceful shutdown", "pid", handle.pid, "error", err)
		}
	} else {
		d.logger.Debug("monitor socket is empty, forcing shutdown")
	}

	// TODO(preetha) we are calling shutdown on the executor here
	// after attempting a graceful qemu shutdown, qemu process may
	// not be around when we call exec.shutdown
	if err := handle.exec.Shutdown(signal, timeout); err != nil {
		if handle.pluginClient.Exited() {
			return nil
		}
		return fmt.Errorf("executor Shutdown failed: %v", err)
	}

	return nil
}


Review comment (refactor):
Can we remove this line because we already have the task handle from line 718?

Code after:

	return ch, nil
}

func (d *Driver) StopTask(taskID string, timeout time.Duration, signal string) error {
	handle, ok := d.tasks.Get(taskID)
	if !ok {
		return drivers.ErrTaskNotFound
	}

	// Attempt a graceful shutdown only if it was configured in the job
	if handle.monitorPath != "" {
		err := sendQemuShutdown(d.logger, handle.monitorPath, handle.pid)
		if err == nil {
			ctx, cancel := context.WithTimeout(context.Background(), timeout)
			defer cancel()
			ticker := time.NewTicker(1 * time.Second)
			defer ticker.Stop()
		out:
			for {
				select {
				case <-ctx.Done():
					d.logger.Error("graceful shutdown", "pid", handle.pid, "timeout after", timeout)
					break out
				case <-ticker.C:
					if handle.procState != drivers.TaskStateRunning {
						break out
					}
				}
			}
		} else {
			d.logger.Debug("error sending graceful shutdown", "pid", handle.pid, "error", err)
		}
	} else {
		d.logger.Debug("monitor socket is empty, forcing shutdown")
	}

	// TODO(preetha) we are calling shutdown on the executor here
	// after attempting a graceful qemu shutdown, qemu process may
	// not be around when we call exec.shutdown
	if err := handle.exec.Shutdown(signal, timeout); err != nil {
		if handle.pluginClient.Exited() {
			return nil
		}
		return fmt.Errorf("executor Shutdown failed: %v", err)
	}

	return nil
}

func (d *Driver) DestroyTask(taskID string, force bool) error {
Repository: hashicorp/nomad
File: nomad/job_endpoint.go (Go)

Code before:

		if args.Eval != nil {
			reply.EvalCreateIndex = index
			reply.EvalID = args.Eval.ID
		}

	} else {
		reply.JobModifyIndex = existingJob.JobModifyIndex
	}

	// used for multiregion start
	args.Job.JobModifyIndex = reply.JobModifyIndex

	// Kick off a multiregion deployment (enterprise only).
	if isRunner {
		// This will skip running deployments for jobs that are not
		// eligible to be run as deployments.
		err = j.multiregionStart(args, reply)
		if err != nil {
			return err
		}
		// We drop any unwanted regions only once we know all jobs have
		// been registered and we've kicked off the deployment (if applicable).
		// This keeps dropping regions close in semantics to dropping task
		// groups in single-region deployments
		err = j.multiregionDrop(args, reply)
		if err != nil {
			return err
		}
	}

	return nil
}

// propagateScalingPolicyIDs propagates scaling policy IDs from existing job
// to updated job, or generates random IDs in new job
func propagateScalingPolicyIDs(old, new *structs.Job) error {

	oldIDs := make(map[string]string)
	if old != nil {
		// use the job-scoped key (includes type, group, and task) to uniquely
		// identify policies in a job
		for _, p := range old.GetScalingPolicies() {
			oldIDs[p.JobKey()] = p.ID
		}
	}

	// ignore any existing ID in the policy, they should be empty
	for _, p := range new.GetScalingPolicies() {
		if id, ok := oldIDs[p.JobKey()]; ok {
			p.ID = id

Review comment (suggestion):
This is all the backcompat logic that can be removed, as it's handled in the job registration raft apply.

Code after:
		}
		return nil
	}

	// Pre-register a deployment if necessary.
	args.Deployment = j.multiregionCreateDeployment(job, args.Eval)
	// Commit this update via Raft
	_, index, err := j.srv.raftApply(structs.JobRegisterRequestType, args)
	if err != nil {
		j.logger.Error("registering job failed", "error", err)
		return err
	}

	// Populate the reply with job information
	reply.JobModifyIndex = index
	reply.Index = index

	if args.Eval != nil {
		reply.EvalCreateIndex = index
	}

	// used for multiregion start
	args.Job.JobModifyIndex = reply.JobModifyIndex

	// Kick off a multiregion deployment (enterprise only).
	if isRunner {
		// This will skip running deployments for jobs that are not
		// eligible to be run as deployments.
		err = j.multiregionStart(args, reply)
		if err != nil {
			return err
		}
		// We drop any unwanted regions only once we know all jobs have
		// been registered and we've kicked off the deployment (if applicable).
		// This keeps dropping regions close in semantics to dropping task
		// groups in single-region deployments
		err = j.multiregionDrop(args, reply)
		if err != nil {
			return err
		}
	}

	return nil
}

// propagateScalingPolicyIDs propagates scaling policy IDs from existing job
// to updated job, or generates random IDs in new job
func propagateScalingPolicyIDs(old, new *structs.Job) error {

	oldIDs := make(map[string]string)
	if old != nil {
Repository: hashicorp/nomad
File: nomad/job_endpoint.go (Go)

Code before:
		// Commit this update via Raft
		_, index, err := j.srv.raftApply(structs.JobRegisterRequestType, args)
		if err != nil {
			j.logger.Error("registering job failed", "error", err)
			return err
		}

		// Populate the reply with job information
		reply.JobModifyIndex = index
		reply.Index = index

		if args.Eval != nil {
			reply.EvalCreateIndex = index
			reply.EvalID = args.Eval.ID
		}

	} else {
		reply.JobModifyIndex = existingJob.JobModifyIndex
	}

	// used for multiregion start
	args.Job.JobModifyIndex = reply.JobModifyIndex

	// Kick off a multiregion deployment (enterprise only).
	if isRunner {
		// This will skip running deployments for jobs that are not
		// eligible to be run as deployments.
		err = j.multiregionStart(args, reply)
		if err != nil {
			return err
		}
		// We drop any unwanted regions only once we know all jobs have
		// been registered and we've kicked off the deployment (if applicable).
		// This keeps dropping regions close in semantics to dropping task
		// groups in single-region deployments
		err = j.multiregionDrop(args, reply)
		if err != nil {
			return err
		}
	}

	return nil
}

// propagateScalingPolicyIDs propagates scaling policy IDs from existing job
// to updated job, or generates random IDs in new job
func propagateScalingPolicyIDs(old, new *structs.Job) error {

	oldIDs := make(map[string]string)
	if old != nil {
		// use the job-scoped key (includes type, group, and task) to uniquely

Review comment (suggestion):
Since we're getting rid of the backcompat logic, we can also get rid of this block as well. It's redundant to the logic at the end of `Job.Register`.

Code after:
			}
			_, evalIndex, err := j.srv.raftApply(structs.EvalUpdateRequestType, update)
			if err != nil {
				j.logger.Error("eval create failed", "error", err, "method", "register")
				return err
			}
			reply.EvalCreateIndex = evalIndex
			reply.Index = evalIndex
		} else {
			reply.Index = existingJob.ModifyIndex
		}
		return nil
	}

	// Pre-register a deployment if necessary.
	args.Deployment = j.multiregionCreateDeployment(job, args.Eval)
	// Commit this update via Raft
	_, index, err := j.srv.raftApply(structs.JobRegisterRequestType, args)
	if err != nil {
		j.logger.Error("registering job failed", "error", err)
		return err
	}

	// Populate the reply with job information
	reply.JobModifyIndex = index
	reply.Index = index

	if args.Eval != nil {
		reply.EvalCreateIndex = index
	}

	// used for multiregion start
	args.Job.JobModifyIndex = reply.JobModifyIndex

	// Kick off a multiregion deployment (enterprise only).
	if isRunner {
		// This will skip running deployments for jobs that are not
		// eligible to be run as deployments.
		err = j.multiregionStart(args, reply)
		if err != nil {
			return err
		}
		// We drop any unwanted regions only once we know all jobs have
		// been registered and we've kicked off the deployment (if applicable).
		// This keeps dropping regions close in semantics to dropping task
		// groups in single-region deployments
		err = j.multiregionDrop(args, reply)
		if err != nil {
			return err
		}
	}
Repository: hashicorp/nomad
File: nomad/job_endpoint.go (Go)

Code before:
			}
			reply.EvalCreateIndex = evalIndex
			reply.Index = evalIndex
		} else {
			reply.Index = existingJob.ModifyIndex
		}
		return nil
	}

	// Pre-register a deployment if necessary.
	args.Deployment = j.multiregionCreateDeployment(job, args.Eval)
	// Commit this update via Raft
	_, index, err := j.srv.raftApply(structs.JobRegisterRequestType, args)
	if err != nil {
		j.logger.Error("registering job failed", "error", err)
		return err
	}

	// Populate the reply with job information
	reply.JobModifyIndex = index
	reply.Index = index

	if args.Eval != nil {
		reply.EvalCreateIndex = index
	}

	// used for multiregion start
	args.Job.JobModifyIndex = reply.JobModifyIndex

	// Kick off a multiregion deployment (enterprise only).
	if isRunner {
		// This will skip running deployments for jobs that are not
		// eligible to be run as deployments.
		err = j.multiregionStart(args, reply)
		if err != nil {
			return err
		}
		// We drop any unwanted regions only once we know all jobs have
		// been registered and we've kicked off the deployment (if applicable).
		// This keeps dropping regions close in semantics to dropping task
		// groups in single-region deployments
		err = j.multiregionDrop(args, reply)
		if err != nil {
			return err
		}
	}

	return nil
}

// propagateScalingPolicyIDs propagates scaling policy IDs from existing job

Review comment (suggestion):
I think we'll always have a created eval here because either it was created for the `!specChanged` case or it was created by the Raft write of the job.

```suggestion
	// Populate the reply with job information
	reply.JobModifyIndex = index
	reply.Index = index
	reply.EvalCreateIndex = index
```

Code after:
				j.logger.Error("eval create failed", "error", err, "method", "register")
				return err
			}
			reply.EvalCreateIndex = evalIndex
			reply.Index = evalIndex
		} else {
			reply.Index = existingJob.ModifyIndex
		}
		return nil
	}

	// Pre-register a deployment if necessary.
	args.Deployment = j.multiregionCreateDeployment(job, args.Eval)
	// Commit this update via Raft
	_, index, err := j.srv.raftApply(structs.JobRegisterRequestType, args)
	if err != nil {
		j.logger.Error("registering job failed", "error", err)
		return err
	}

	// Populate the reply with job information
	reply.JobModifyIndex = index
	reply.Index = index

	if args.Eval != nil {
		reply.EvalCreateIndex = index
	}

	// used for multiregion start
	args.Job.JobModifyIndex = reply.JobModifyIndex

	// Kick off a multiregion deployment (enterprise only).
	if isRunner {
		// This will skip running deployments for jobs that are not
		// eligible to be run as deployments.
		err = j.multiregionStart(args, reply)
		if err != nil {
			return err
		}
		// We drop any unwanted regions only once we know all jobs have
		// been registered and we've kicked off the deployment (if applicable).
		// This keeps dropping regions close in semantics to dropping task
		// groups in single-region deployments
		err = j.multiregionDrop(args, reply)
		if err != nil {
			return err
		}
	}

	return nil
}
Repository: hashicorp/nomad
File: enos/modules/test_cluster_health/scripts/allocs.sh (Shell)

Code before:
#!/usr/bin/env bash
# Copyright IBM Corp. 2015, 2025
# SPDX-License-Identifier: BUSL-1.1

set -euo pipefail

error_exit() {
    printf 'Error: %s\n' "${1}"

    ALL_ALLOCS=$(nomad alloc status -json)
    echo "$ALL_ALLOCS" > /tmp/allocs.json
    cat /tmp/allocs.json | jq -r '
        ["ID", "Node", "ClientStatus", "DesiredStatus", "JobID"],
        ["--------", "--------", "------------", "-------------", "---------------"],
        (.[] | [.ID[:8], .NodeID[:8], .ClientStatus, .DesiredStatus, .JobID])
        | @tsv' | column -ts $'\t'

    echo "full allocation status for debugging written to: /tmp/allocs.json"
    exit 1
}

MAX_WAIT_TIME=120
POLL_INTERVAL=2

elapsed_time=0

# Quality: nomad_allocs_status: A GET call to /v1/allocs returns the correct number of allocations and they are all running

running_allocs=
allocs_length=
last_error=

checkAllocsCount() {
    local allocs
    allocs=$(nomad alloc status -json) || {
        last_error="Failed to check alloc status"

Review comment (suggestion):
As I mentioned in [the e2e repo PR](https://github.com/hashicorp/nomad-e2e/pull/175), I think it'd be nice to settle on `/tmp/artifacts/` as a target for files like this, then we can count on that when attaching them to a GHA run.

Code after:
#!/usr/bin/env bash
# Copyright IBM Corp. 2015, 2025
# SPDX-License-Identifier: BUSL-1.1

set -euo pipefail

error_exit() {
    printf 'Error: %s\n' "${1}"
    ALL_ALLOCS=$(nomad alloc status -json)
    mkdir -p /tmp/artifacts
    OUT="/tmp/artifacts/allocs.json"
    echo "$ALL_ALLOCS" > "$OUT"

    cat "$OUT" | jq -r '
        ["ID", "Node", "ClientStatus", "DesiredStatus", "JobID"],
        ["--------", "--------", "------------", "-------------", "---------------"],
        (.[] | [.ID[:8], .NodeID[:8], .ClientStatus, .DesiredStatus, .JobID])
        | @tsv' | column -ts $'\t'

    echo "full allocation status for debugging written to: $OUT"
    exit 1
}

MAX_WAIT_TIME=120
POLL_INTERVAL=2

elapsed_time=0

# Quality: nomad_allocs_status: A GET call to /v1/allocs returns the correct number of allocations and they are all running

running_allocs=
allocs_length=
last_error=

checkAllocsCount() {
    local allocs
Repository: hashicorp/nomad
File: nomad/plan_apply_test.go (Go)

Code before:

	fit, reason, err := evaluateNodePlan(snap, plan, node.ID)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	if !fit {
		t.Fatalf("bad")
	}
	if reason != "" {
		t.Fatalf("bad")
	}
}

func TestPlanApply_EvalNodePlan_NodeDown_EvictOnly(t *testing.T) {
	ci.Parallel(t)
	alloc := mock.Alloc()
	state := testStateStore(t)
	node := mock.Node()
	alloc.NodeID = node.ID
	alloc.AllocatedResources = structs.NodeResourcesToAllocatedResources(node.NodeResources)
	node.ReservedResources = nil
	node.Status = structs.NodeStatusDown
	state.UpsertNode(structs.MsgTypeTestSetup, 1000, node)
	state.UpsertAllocs(structs.MsgTypeTestSetup, 1001, []*structs.Allocation{alloc})
	snap, _ := state.Snapshot()

	allocEvict := new(structs.Allocation)
	*allocEvict = *alloc
	allocEvict.DesiredStatus = structs.AllocDesiredStatusEvict
	plan := &structs.Plan{
		JobInfo: &structs.PlanJobTuple{
			Namespace: alloc.Job.Namespace,
			ID:        alloc.Job.ID,
			Version:   alloc.Job.Version,
		},
		NodeUpdate: map[string][]*structs.Allocation{
			node.ID: {allocEvict},
		},
	}

	fit, reason, err := evaluateNodePlan(snap, plan, node.ID)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	if !fit {
		t.Fatalf("bad")
	}
	if reason != "" {
		t.Fatalf("bad")
	}
}

Review comment (bug):
This test case points out a weird edge case I hadn't considered in the original issue: if the plan _only_ contains older allocations being stopped, which version should we use?

Code after:
	if reason != "" {
		t.Fatalf("bad")
	}
}

func TestPlanApply_EvalNodePlan_NodeDown_EvictOnly(t *testing.T) {
	ci.Parallel(t)
	alloc := mock.Alloc()
	state := testStateStore(t)
	node := mock.Node()
	alloc.NodeID = node.ID
	alloc.AllocatedResources = structs.NodeResourcesToAllocatedResources(node.NodeResources)
	node.ReservedResources = nil
	node.Status = structs.NodeStatusDown
	state.UpsertNode(structs.MsgTypeTestSetup, 1000, node)
	state.UpsertAllocs(structs.MsgTypeTestSetup, 1001, []*structs.Allocation{alloc})
	snap, _ := state.Snapshot()

	allocEvict := new(structs.Allocation)
	*allocEvict = *alloc
	allocEvict.DesiredStatus = structs.AllocDesiredStatusEvict
	plan := &structs.Plan{
		JobInfo: &structs.PlanJobTuple{
			Namespace: alloc.Job.Namespace,
			ID:        alloc.Job.ID,
		},
		NodeUpdate: map[string][]*structs.Allocation{
			node.ID: {allocEvict},
		},
	}

	fit, reason, err := evaluateNodePlan(snap, plan, node.ID)
	if err != nil {
		t.Fatalf("err: %v", err)
	}
	if !fit {
		t.Fatalf("bad")
	}
	if reason != "" {
		t.Fatalf("bad")
	}
}

// TestPlanApply_EvalNodePlan_Node_Disconnected tests that plans for disconnected
// nodes can only contain allocs with client status unknown.
func TestPlanApply_EvalNodePlan_Node_Disconnected(t *testing.T) {
	ci.Parallel(t)

	state := testStateStore(t)
	node := mock.Node()
	node.Status = structs.NodeStatusDisconnected
Repository: hashicorp/nomad
File: command/agent/event_endpoint.go (Go)

Code before:

			// If there is a buffered reader/writer, flush it to ensure headers
			// are sent.
			if bufrw != nil {
				_ = bufrw.Flush()
			}
			output = &deadlineWriter{conn: conn, timeout: writeTimeout}
		}
	}
	if output == nil {
		// Fallback: the existing flusher (no-op Close).
		output = &nopCloser{w: ioutils.NewWriteFlusher(resp)}
	}

	// Create a goroutine that closes the pipe if the connection closes
	go func() {
		<-ctx.Done()
		httpPipe.Close()
		output.Close()
	}()

	// send request and decode events
	errs, errCtx := errgroup.WithContext(ctx)
	errs.Go(func() error {
		defer cancel()

		// Send the request
		if err := encoder.Encode(args); err != nil {
			return CodedError(500, err.Error())
		}

		for {
			select {
			case <-errCtx.Done():
				return nil
			default:
			}

			// Decode the response
			var res structs.EventStreamWrapper
			if err := decoder.Decode(&res); err != nil {
				return CodedError(500, err.Error())
			}
			decoder.Reset(httpPipe)

			if err := res.Error; err != nil {
				if err.Code != nil {
					return CodedError(int(*err.Code), err.Error())
				}
			}


Review comment (bug):
Looking at [the docs](https://pkg.go.dev/net/http#Hijacker):
> The default ResponseWriter for HTTP/1.x connections supports Hijacker, but HTTP/2 connections intentionally do not. ResponseWriter wrappers may also not support Hijacker. Handlers should always test for this ability at runtime.

It seems this code path gets triggered only if we haven't wrapped the response writer in something that implements `Hijack` or the `Hijack` method itself fails. Does our `(*HTTPServer) wrap` method hit this case?

Code after:

			res := &http.Response{
				Status:        "200 OK",
				StatusCode:    http.StatusOK,
				Header:        headers,
				ContentLength: -1,
			}

			_ = conn.SetWriteDeadline(time.Now().Add(writeTimeout))
			_ = res.Write(conn)

			// If the hijack returned a buffered writer/reader, flush any
			// buffered data so middleware or the net/http stack doesn't
			// leave pending bytes unflushed on the connection. Flush after
			// writing headers to preserve header ordering.
			if bufrw != nil {
				_ = bufrw.Flush()
			}

			output = ioutils.NewWriteFlusher(&deadlineWriter{conn: conn, timeout: writeTimeout})
		}
	}
	if output == nil {
		// Fallback: the existing flusher (no-op Close).
		output = ioutils.NewWriteFlusher(resp)
	}

	// Create a goroutine that closes the pipe if the connection closes
	go func() {
		<-ctx.Done()
		httpPipe.Close()
		output.Close()
	}()

	// send request and decode events
	errs, errCtx := errgroup.WithContext(ctx)
	errs.Go(func() error {
		defer cancel()

		// Send the request
		if err := encoder.Encode(args); err != nil {
			return CodedError(500, err.Error())
		}

		for {
			select {
			case <-errCtx.Done():
				return nil
			default:
			}