245 lines
11 KiB
Go
245 lines
11 KiB
Go
package api
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/dto"
|
|
"browser.local/platform/validator"
|
|
)
|
|
|
|
func TestRunChannelAPIInterleavedRequestsMutateIndependentState(t *testing.T) {
|
|
router := newTestRouter()
|
|
hello := createArtifactTransferAPIFixtures(t, router)
|
|
createOnlyLogStreamForChannelIsolation(t, router)
|
|
|
|
claimRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/claim", dto.RunJobClaimRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: hello.SessionToken,
|
|
Capabilities: []string{"process.start"},
|
|
Capacity: dto.RunCapacityResponse{MaxJobs: 4},
|
|
})
|
|
assertStatus(t, claimRecorder, http.StatusOK)
|
|
claim := decodeBody[dto.RunJobClaimResponse](t, claimRecorder)
|
|
if !claim.HasJob || claim.Job.JobID != "job-1" {
|
|
t.Fatalf("expected claimed job, got %+v", claim)
|
|
}
|
|
|
|
payload := []byte("interleaved artifact payload for api isolation")
|
|
openRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/open", artifactOpenForChannelIsolation(hello.SessionToken, payload, 8))
|
|
assertStatus(t, openRecorder, http.StatusOK)
|
|
open := decodeBody[dto.ArtifactTransferOpenResponse](t, openRecorder)
|
|
firstChunkRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/chunks", artifactChunkForChannelIsolation(hello.SessionToken, open.TransferID, payload, 0, 8))
|
|
assertStatus(t, firstChunkRecorder, http.StatusOK)
|
|
firstChunk := decodeBody[dto.ArtifactChunkUploadResponse](t, firstChunkRecorder)
|
|
if !firstChunk.Accepted || firstChunk.NextMissingChunkIndex != 1 {
|
|
t.Fatalf("expected first artifact chunk ack, got %+v", firstChunk)
|
|
}
|
|
|
|
heartbeatRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/control/heartbeat", dto.RunControlHeartbeatRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: hello.SessionToken,
|
|
Version: "0.1.1",
|
|
Status: domain.RunEndpointStatusOnline,
|
|
CapabilityFingerprint: "cap-artifacts",
|
|
Capacity: dto.RunCapacityResponse{MaxJobs: 4, RunningJobs: 1},
|
|
})
|
|
assertStatus(t, heartbeatRecorder, http.StatusOK)
|
|
heartbeat := decodeBody[dto.RunControlHeartbeatResponse](t, heartbeatRecorder)
|
|
if !heartbeat.Accepted {
|
|
t.Fatalf("expected heartbeat accepted, got %+v", heartbeat)
|
|
}
|
|
|
|
ack := postRunJobAck(t, router, dto.RunJobAckRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: hello.SessionToken,
|
|
JobID: claim.Job.JobID,
|
|
LeaseToken: claim.Job.LeaseToken,
|
|
Attempt: claim.Job.Attempt,
|
|
Message: "job accepted while artifact transfer is active",
|
|
})
|
|
if ack.Job.State != domain.JobStateRunning {
|
|
t.Fatalf("expected running job after ack, got %+v", ack)
|
|
}
|
|
|
|
logBatch := validLogBatchRequest(t, hello.SessionToken, 1, 1)
|
|
logBatch.LogStreamID = "log-channel-isolation"
|
|
logBatch.LogSessionID = ""
|
|
logBatch.SessionStartedAt = time.Time{}
|
|
logRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/logs/batches", logBatch)
|
|
assertStatus(t, logRecorder, http.StatusOK)
|
|
logAck := decodeBody[dto.LogBatchIngestResponse](t, logRecorder)
|
|
if !logAck.Accepted || logAck.LatestSeq != 1 {
|
|
t.Fatalf("expected log ack independent from artifact transfer, got %+v", logAck)
|
|
}
|
|
|
|
resultRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/result", dto.RunJobResultRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: hello.SessionToken,
|
|
JobID: claim.Job.JobID,
|
|
LeaseToken: claim.Job.LeaseToken,
|
|
Attempt: claim.Job.Attempt,
|
|
State: domain.JobStateSucceeded,
|
|
Progress: dto.JobProgressBody{Percent: 100, Message: "done"},
|
|
ResultRef: "artifact://jobs/job-1/result",
|
|
Message: "done",
|
|
})
|
|
assertStatus(t, resultRecorder, http.StatusOK)
|
|
result := decodeBody[dto.RunJobResultResponse](t, resultRecorder)
|
|
if result.Job.State != domain.JobStateSucceeded || result.Job.ResultRef == "" {
|
|
t.Fatalf("expected terminal result independent from transfer, got %+v", result)
|
|
}
|
|
|
|
statusRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/artifacts/status", dto.ArtifactTransferStatusRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: hello.SessionToken,
|
|
TransferID: open.TransferID,
|
|
ArtifactID: "artifact-channel-isolation",
|
|
})
|
|
assertStatus(t, statusRecorder, http.StatusOK)
|
|
status := decodeBody[dto.ArtifactTransferStatusResponse](t, statusRecorder)
|
|
if status.Completed || status.NextMissingChunkIndex != 1 || len(status.ReceivedChunkIndexes) != 1 {
|
|
t.Fatalf("artifact state should remain independent after heartbeat/job/log calls, got %+v", status)
|
|
}
|
|
stream := getJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams/log-channel-isolation")
|
|
if stream.LatestSeq != 1 {
|
|
t.Fatalf("expected log stream latest seq updated independently, got %+v", stream)
|
|
}
|
|
job := getJSON[dto.JobResponse](t, router, "/api/v1/jobs/job-1")
|
|
if job.State != domain.JobStateSucceeded {
|
|
t.Fatalf("expected job terminal state preserved, got %+v", job)
|
|
}
|
|
}
|
|
|
|
func TestLightweightRunRoutesRejectHeavyChannelPayloads(t *testing.T) {
|
|
router := newTestRouter()
|
|
hello := createArtifactTransferAPIFixtures(t, router)
|
|
createOnlyLogStreamForChannelIsolation(t, router)
|
|
postJSON[dto.JobResponse](t, router, "/api/v1/jobs", dto.JobCreateRequest{
|
|
ID: "job-heavy-payload",
|
|
ServerInstanceID: "server-1",
|
|
RunEndpointID: "run-local",
|
|
Capability: "process.start",
|
|
IdempotencyKey: "heavy-payload-job",
|
|
})
|
|
claimRecorder := performJSON(t, router, http.MethodPost, "/api/v1/run/jobs/claim", dto.RunJobClaimRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: hello.SessionToken,
|
|
Capabilities: []string{"process.start"},
|
|
Capacity: dto.RunCapacityResponse{MaxJobs: 4},
|
|
})
|
|
assertStatus(t, claimRecorder, http.StatusOK)
|
|
claim := decodeBody[dto.RunJobClaimResponse](t, claimRecorder)
|
|
|
|
for _, tc := range []struct {
|
|
name string
|
|
path string
|
|
body string
|
|
}{
|
|
{
|
|
name: "heartbeat rejects artifact chunk fields",
|
|
path: "/api/v1/run/control/heartbeat",
|
|
body: fmt.Sprintf(`{"runEndpointId":"run-local","sessionToken":%q,"version":"0.1.1","status":"online","capabilityFingerprint":"cap-jobs","capacity":{"maxJobs":4},"payload":"AAAA","transferId":"transfer-1","hostPath":"/Users/tasia/server"}`, hello.SessionToken),
|
|
},
|
|
{
|
|
name: "job result rejects inline logs and sockets",
|
|
path: "/api/v1/run/jobs/result",
|
|
body: fmt.Sprintf(`{"runEndpointId":"run-local","sessionToken":%q,"jobId":%q,"leaseToken":%q,"attempt":%d,"state":"succeeded","progress":{"percent":100},"resultRef":"artifact://jobs/job-heavy-payload/result","entries":[{"seq":1,"line":"log"}],"directSocket":"unix:///tmp/run.sock"}`, hello.SessionToken, claim.Job.JobID, claim.Job.LeaseToken, claim.Job.Attempt),
|
|
},
|
|
{
|
|
name: "log ingest rejects artifact transfer payload",
|
|
path: "/api/v1/run/logs/batches",
|
|
body: fmt.Sprintf(`{"runEndpointId":"run-local","sessionToken":%q,"logStreamId":"log-channel-isolation","serverInstanceId":"server-1","streamKey":"stdout","source":"process","firstSeq":1,"lastSeq":1,"compression":"none","checksum":"sha256:bad","entries":[{"seq":1,"timestamp":"2026-07-03T12:00:01Z","line":"line"}],"payload":"AAAA","transferId":"transfer-1"}`, hello.SessionToken),
|
|
},
|
|
} {
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
recorder := performRaw(t, router, http.MethodPost, tc.path, tc.body)
|
|
assertErrorResponse(t, recorder, http.StatusBadRequest, errorCodeBadRequest)
|
|
})
|
|
}
|
|
}
|
|
|
|
func createLogStreamForChannelIsolation(t *testing.T, router http.Handler) {
|
|
t.Helper()
|
|
adminSession := createAdminSession(t, router)
|
|
postJSON[dto.GamePluginResponse](t, router, "/api/v1/game-plugins", validGamePluginRequest())
|
|
postJSONWithAuth[dto.ServerInstanceResponse](t, router, "/api/v1/server-instances", dto.ServerInstanceCreateRequest{
|
|
ID: "server-1",
|
|
PluginID: "server.scum",
|
|
RunEndpointID: "run-local",
|
|
Name: "SCUM #1",
|
|
}, adminSession)
|
|
postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{
|
|
ID: "log-channel-isolation",
|
|
ServerInstanceID: "server-1",
|
|
Source: domain.LogStreamSourceProcess,
|
|
StreamKey: "stdout",
|
|
StorageBackend: domain.LogStorageBackendLocalSegments,
|
|
RetentionPolicy: "default",
|
|
})
|
|
}
|
|
|
|
func createOnlyLogStreamForChannelIsolation(t *testing.T, router http.Handler) {
|
|
t.Helper()
|
|
postJSON[dto.LogStreamResponse](t, router, "/api/v1/log-streams", dto.LogStreamCreateRequest{
|
|
ID: "log-channel-isolation",
|
|
ServerInstanceID: "server-1",
|
|
Source: domain.LogStreamSourceProcess,
|
|
StreamKey: "stdout",
|
|
StorageBackend: domain.LogStorageBackendLocalSegments,
|
|
RetentionPolicy: "default",
|
|
})
|
|
}
|
|
|
|
func artifactOpenForChannelIsolation(sessionToken string, payload []byte, chunkSize int) dto.ArtifactTransferOpenRequest {
|
|
return dto.ArtifactTransferOpenRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: sessionToken,
|
|
ArtifactID: "artifact-channel-isolation",
|
|
Direction: domain.ArtifactTransferDirectionUpload,
|
|
OwnerKind: domain.ArtifactOwnerKindJob,
|
|
OwnerID: "job-1",
|
|
SizeBytes: int64(len(payload)),
|
|
ChunkSizeBytes: chunkSize,
|
|
Checksum: validator.BytesChecksum(payload),
|
|
IdempotencyKey: "artifact-channel-isolation",
|
|
}
|
|
}
|
|
|
|
func artifactChunkForChannelIsolation(sessionToken string, transferID string, payload []byte, index int, chunkSize int) dto.ArtifactChunkUploadRequest {
|
|
offset := index * chunkSize
|
|
end := offset + chunkSize
|
|
if end > len(payload) {
|
|
end = len(payload)
|
|
}
|
|
part := payload[offset:end]
|
|
return dto.ArtifactChunkUploadRequest{
|
|
RunEndpointID: "run-local",
|
|
SessionToken: sessionToken,
|
|
TransferID: transferID,
|
|
ArtifactID: "artifact-channel-isolation",
|
|
ChunkIndex: index,
|
|
Offset: int64(offset),
|
|
SizeBytes: len(part),
|
|
Checksum: validator.BytesChecksum(part),
|
|
Payload: part,
|
|
}
|
|
}
|
|
|
|
func TestRunChannelAPIHeavyPayloadRejectionsDoNotMutateState(t *testing.T) {
|
|
router := newTestRouter()
|
|
hello := decodeBody[dto.RunControlHelloResponse](t, performRunControlHello(t, router, validRunJobControlHelloRequest()))
|
|
|
|
recorder := performRaw(t, router, http.MethodPost, "/api/v1/run/control/heartbeat", fmt.Sprintf(`{"runEndpointId":"run-local","sessionToken":%q,"version":"0.1.1","status":"online","capabilityFingerprint":"cap-jobs","capacity":{"maxJobs":4,"runningJobs":1},"payload":"AAAA"}`, hello.SessionToken))
|
|
assertErrorResponse(t, recorder, http.StatusBadRequest, errorCodeBadRequest)
|
|
|
|
endpoint := getJSON[dto.RunEndpointResponse](t, router, "/api/v1/run/endpoints/run-local")
|
|
if endpoint.Capacity.RunningJobs != 0 || strings.Contains(endpoint.Capacity.Summary, "AAAA") {
|
|
t.Fatalf("rejected heartbeat must not mutate endpoint capacity or store heavy payload, got %+v", endpoint)
|
|
}
|
|
}
|