fix(server): include nodes and creds in request dedupe

pkg/server/http_server.go

@@ -204,6 +204,8 @@ func readRequest(w http.ResponseWriter, r *http.Request) *topology.Request {
 		}
 	}
 
+	tr.Provider.Creds = checkCredentials(tr.Provider.Creds, srv.cfg.Credentials)
+
 	klog.Info(tr.String())
 
 	if err = validate(tr); err != nil {

pkg/server/http_server_test.go

@@ -621,6 +621,61 @@ func TestReadRequest(t *testing.T) {
 	}
 }
 
+func TestReadRequestAppliesEffectiveCredentials(t *testing.T) {
+	configCreds := map[string]any{"token": "config-token"}
+	payloadCreds := map[string]any{"token": "payload-token"}
+
+	srv = &HttpServer{
+		cfg: &config.Config{
+			Provider:    "test",
+			Engine:      "slurm",
+			Credentials: configCreds,
+		},
+	}
+
+	testCases := []struct {
+		name     string
+		payload  string
+		expected map[string]any
+	}{
+		{
+			name:     "Test readRequest with config credentials",
+			payload:  fmt.Sprintf(simpleSlurmPayload, "test"),
+			expected: configCreds,
+		},
+		{
+			name: "Test readRequest with payload credentials",
+			payload: `{
+			  "provider": {
+			    "name": "test",
+			    "creds": {
+			      "token": "payload-token"
+			    }
+			  },
+			  "engine": {
+			    "name": "slurm"
+			  }
+			}`,
+			expected: payloadCreds,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			r := &http.Request{
+				Method: http.MethodPost,
+				Body:   io.NopCloser(bytes.NewBuffer([]byte(tc.payload))),
+			}
+
+			w := httptest.NewRecorder()
+			req := readRequest(w, r)
+
+			require.NotNil(t, req)
+			require.Equal(t, tc.expected, req.Provider.Creds)
+		})
+	}
+}
+
 func readInvalidRequest(t *testing.T, payload, msg string) {
 	r := &http.Request{
 		Method: http.MethodPost,

pkg/server/trailing_delay_queue.go

@@ -122,7 +122,8 @@ func (q *TrailingDelayQueue) Get(hash string) *Completion {
 	defer q.mutex.Unlock()
 
 	if res, ok := q.store.Get(hash); ok {
-		return res.(*Completion)
+		completion := *(res.(*Completion))
+		return &completion
 	}
 
 	return &Completion{

pkg/server/trailing_delay_queue_test.go

@@ -88,6 +88,79 @@ func TestVaryingPayload(t *testing.T) {
 	queue.Shutdown()
 }
 
+func TestVaryingPayloadByNodesAndCredentials(t *testing.T) {
+	processItem := func(item any) (any, *httperr.Error) {
+		return item, nil
+	}
+
+	queue := NewTrailingDelayQueue(processItem, 10*time.Millisecond)
+	defer queue.Shutdown()
+
+	requests := []*topology.Request{
+		{
+			Provider: topology.Provider{
+				Name:  "test",
+				Creds: map[string]any{"token": "a"},
+			},
+			Engine: topology.Engine{Name: "slurm"},
+			Nodes: []topology.ComputeInstances{
+				{
+					Region:    "region",
+					Instances: map[string]string{"instance-1": "node-1"},
+				},
+			},
+		},
+		{
+			Provider: topology.Provider{
+				Name:  "test",
+				Creds: map[string]any{"token": "a"},
+			},
+			Engine: topology.Engine{Name: "slurm"},
+			Nodes: []topology.ComputeInstances{
+				{
+					Region:    "region",
+					Instances: map[string]string{"instance-2": "node-2"},
+				},
+			},
+		},
+		{
+			Provider: topology.Provider{
+				Name:  "test",
+				Creds: map[string]any{"token": "b"},
+			},
+			Engine: topology.Engine{Name: "slurm"},
+			Nodes: []topology.ComputeInstances{
+				{
+					Region:    "region",
+					Instances: map[string]string{"instance-1": "node-1"},
+				},
+			},
+		},
+	}
+
+	submissions := make([]string, 0, len(requests))
+	for _, request := range requests {
+		uid, err := queue.Submit(request)
+		require.NoError(t, err)
+		submissions = append(submissions, uid)
+	}
+
+	for i := 1; i < len(submissions); i++ {
+		require.NotEqual(t, submissions[i], submissions[i-1])
+	}
+	require.NotEqual(t, submissions[0], submissions[2])
+
+	require.Eventually(t, func() bool {
+		for i, uid := range submissions {
+			res := queue.Get(uid)
+			if res.Status != http.StatusOK || res.Ret != requests[i] {
+				return false
+			}
+		}
+		return true
+	}, time.Second, 10*time.Millisecond)
+}
+
 func TestLRU(t *testing.T) {
 	cache, _ := lru.New(3)
 

pkg/topology/request.go

@@ -17,11 +17,22 @@
 package topology
 
 import (
+	"crypto/hmac"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"hash/fnv"
 	"sort"
 	"strings"
+	"sync"
+)
+
+var (
+	credentialHashKey     []byte
+	credentialHashKeyErr  error
+	credentialHashKeyOnce sync.Once
 )
 
 type Request struct {
@@ -46,6 +57,28 @@ type ComputeInstances struct {
 	Instances map[string]string `json:"instances"` // <instance ID>:<node name> map
 }
 
+type requestHashData struct {
+	Provider providerHashData       `json:"provider"`
+	Engine   Engine                 `json:"engine"`
+	Nodes    []computeInstancesHash `json:"nodes,omitempty"`
+}
+
+type providerHashData struct {
+	Name             string         `json:"name"`
+	Params           map[string]any `json:"params,omitempty"`
+	CredentialDigest string         `json:"credentialDigest,omitempty"`
+}
+
+type computeInstancesHash struct {
+	Region    string         `json:"region"`
+	Instances []instanceHash `json:"instances"`
+}
+
+type instanceHash struct {
+	ID   string `json:"id"`
+	Node string `json:"node"`
+}
+
 func NewRequest(prv Provider, eng Engine) *Request {
 	return &Request{
 		Provider: prv,
@@ -141,19 +174,112 @@ func GetNodeNameMap(cis []ComputeInstances) map[string]bool {
 }
 
 func (p *Request) Hash() (string, error) {
-	dataToHash := Request{
-		Provider: Provider{
-			Name:   p.Provider.Name,
-			Params: p.Provider.Params,
+	credentialDigest, err := getCredentialDigest(p.Provider.Creds)
+	if err != nil {
+		return "", err
+	}
+
+	dataToHash := requestHashData{
+		Provider: providerHashData{
+			Name:             p.Provider.Name,
+			Params:           p.Provider.Params,
+			CredentialDigest: credentialDigest,
 		},
 		Engine: Engine{
 			Name:   p.Engine.Name,
 			Params: p.Engine.Params,
 		},
+		Nodes: canonicalComputeInstances(p.Nodes),
 	}
 	return GetHash(dataToHash)
 }
 
+func canonicalComputeInstances(nodes []ComputeInstances) []computeInstancesHash {
+	if len(nodes) == 0 {
+		return nil
+	}
+
+	canonical := make([]computeInstancesHash, 0, len(nodes))
+	for _, nodeGroup := range nodes {
+		instances := make([]instanceHash, 0, len(nodeGroup.Instances))
+		for id, node := range nodeGroup.Instances {
+			instances = append(instances, instanceHash{
+				ID:   id,
+				Node: node,
+			})
+		}
+		sort.Slice(instances, func(i, j int) bool {
+			if instances[i].ID != instances[j].ID {
+				return instances[i].ID < instances[j].ID
+			}
+			return instances[i].Node < instances[j].Node
+		})
+
+		canonical = append(canonical, computeInstancesHash{
+			Region:    nodeGroup.Region,
+			Instances: instances,
+		})
+	}
+
+	sort.Slice(canonical, func(i, j int) bool {
+		if canonical[i].Region != canonical[j].Region {
+			return canonical[i].Region < canonical[j].Region
+		}
+		if len(canonical[i].Instances) != len(canonical[j].Instances) {
+			return len(canonical[i].Instances) < len(canonical[j].Instances)
+		}
+		for idx := range canonical[i].Instances {
+			if canonical[i].Instances[idx].ID != canonical[j].Instances[idx].ID {
+				return canonical[i].Instances[idx].ID < canonical[j].Instances[idx].ID
+			}
+			if canonical[i].Instances[idx].Node != canonical[j].Instances[idx].Node {
+				return canonical[i].Instances[idx].Node < canonical[j].Instances[idx].Node
+			}
+		}
+		return false
+	})
+
+	return canonical
+}
+
+func getCredentialDigest(creds map[string]any) (string, error) {
+	if len(creds) == 0 {
+		return "", nil
+	}
+
+	data, err := json.Marshal(creds)
+	if err != nil {
+		return "", fmt.Errorf("failed to marshal credentials for hashing: %v", err)
+	}
+
+	key, err := getCredentialHashKey()
+	if err != nil {
+		return "", err
+	}
+
+	mac := hmac.New(sha256.New, key)
+	_, _ = mac.Write(data)
+	return hex.EncodeToString(mac.Sum(nil)), nil
+}
+
+func getCredentialHashKey() ([]byte, error) {
+	credentialHashKeyOnce.Do(func() {
+		credentialHashKey, credentialHashKeyErr = newCredentialHashKey()
+	})
+	if credentialHashKeyErr != nil {
+		return nil, credentialHashKeyErr
+	}
+	return credentialHashKey, nil
+}
+
+func newCredentialHashKey() ([]byte, error) {
+	key := make([]byte, sha256.Size)
+	if _, err := rand.Read(key); err != nil {
+		return nil, fmt.Errorf("failed to generate credential hash key: %v", err)
+	}
+	return key, nil
+}
+
 func GetHash(obj any) (string, error) {
 	data, err := json.Marshal(obj)
 	if err != nil {