fix: 10k concurrency stress tests and fixes

[alexlovelltroy]

Description

This pull request introduces a new asynchronous, backpressured queue for removing WireGuard peers, refactors the PhoneHomeHandler to use this queue, and adds comprehensive stress and unit tests for the queue and for defensive copying in MemStore. Additionally, it ensures that all data returned from MemStore is defensively copied to prevent concurrent mutation bugs. The release workflow is updated to run new stress tests.

WireGuard Peer Removal Queue:

• Added PeerRemovalQueue, an asynchronous, backpressured queue for removing WireGuard peers, replacing direct removal calls with a bounded worker pool to prevent resource exhaustion and provide backpressure if the queue is full. (cmd/cloud-init-server/peer_removal_queue.go)
• Refactored PhoneHomeHandler to use PeerRemovalQueue instead of directly calling the WireGuard interface manager, returning HTTP 503 if the queue is full. (cmd/cloud-init-server/handlers.go, cmd/cloud-init-server/main.go) [1] [2] [3] [4] [5] [6]

Testing and CI Improvements:

• Added unit and stress tests for PeerRemovalQueue to validate queue bounds, concurrency, and handler responses under load. (cmd/cloud-init-server/peer_removal_queue_test.go, cmd/cloud-init-server/peer_removal_queue_stress_test.go) [1] [2]
• Updated the GitHub Actions release workflow to run new stress tests before releasing. (.github/workflows/Release.yml)

Defensive Copying in MemStore:

• Modified all relevant MemStore methods to return deep defensive copies of stored data, preventing concurrent mutation and data races. New helper functions perform deep copying of maps, slices, and nested structures. (internal/memstore/ciMemStore.go) [1] [2] [3] [4] [5] [6] [7] [8]
• Added a stress test to verify that concurrent access and mutation of returned data from MemStore does not affect the underlying store. (internal/memstore/ciMemStore_stress_test.go)

Dependency and Import Cleanups:

• Removed unused imports and updated import statements as needed to support the above changes. (cmd/cloud-init-server/handlers.go, internal/memstore/ciMemStore.go) [1] [2]

Checklist

• My code follows the style guidelines of this project
• I have added/updated comments where needed
• I have added tests that prove my fix is effective or my feature works
• I have run make test (or equivalent) locally and all tests pass
• I have updated the relevant documentation (CLI examples, man pages, README, other docs, etc.)
• DCO Sign-off: All commits are signed off (git commit -s) with my real name and email
• REUSE Compliance:
  • Each new/modified source file has SPDX copyright and license headers
  • Any non-commentable files include a <filename>.license sidecar
  • All referenced licenses are present in the LICENSES/ directory

Type of Change

• Bug fix
• New feature
• Breaking change
• Documentation update
• Dependency update

---

For more info, see Contributing Guidelines.

[travisbcotton]

unrelated to the PR but this looks like it might call close twice

[travisbcotton]

Looks ok to me after the IP release fix

[synackd]

One potential issue I can see is that, since the peer removal queue only stores the name[1], [2], RemovePeer() looks up the current peer key to remove. If the node reconnects while removal is queued, IpForPeer() replaces that key, and the stale removal job removes the new WireGuard session and deletes its map entry.

To mitigate, we could include both the peer name and peer key (captured at enqueue time) in the removal job.

[synackd]

Do we want to add ?type=Node to this endpoint so we don't copy any unnecessary components into memory?

[synackd]

refreshTokenIfCurrent() uses the same mutex (s.accessTokenMutex) as s.currentAccessToken() above. The former locks the mutex while making the HTTP call to make the token refresh request and doesn't seem to use a timeout:

cloud-init/internal/smdclient/oidc.go

Lines 27 to 55 in bd7e138

	func (s *SMDClient) refreshTokenIfCurrent(rejectedToken string) error {
	s.accessTokenMutex.Lock()
	defer s.accessTokenMutex.Unlock()
	if s.accessToken != rejectedToken {
	return nil
	}
	return s.refreshTokenLocked()
	}
	func (s *SMDClient) refreshTokenLocked() error {
	// Request new token from OIDC server
	r, err := http.Get(s.tokenEndpoint)
	if err != nil {
	return err
	}
	defer r.Body.Close()
	body, err := io.ReadAll(r.Body)
	if err != nil {
	return err
	}
	// Decode server's response to the expected structure
	var tokenResp oidcTokenData
	if err = json.Unmarshal(body, &tokenResp); err != nil {
	return err
	}
	// Extract and store the JWT itself
	s.accessToken = tokenResp.Access_token
	return nil
	}

I might be worried that this will stall all other SMD operations while the mutex is held indefinitely.

I wonder if it would be good to implement timeout/retry logic here to prevent indefinite deadlock.

[synackd]

"here" above == linked code block, not the comment lines.