Fix server file browse flow
This commit is contained in:
@@ -95,6 +95,7 @@ func (h *coreHandlers) register(mux *http.ServeMux) {
|
||||
mux.HandleFunc("/api/v1/server-instances/{id}/logs/events", h.serverLogEvents)
|
||||
mux.HandleFunc("/api/v1/server-instances/{id}/files/workspace", h.serverFilesWorkspace)
|
||||
mux.HandleFunc("/api/v1/server-instances/{id}/files/list", h.serverFilesList)
|
||||
mux.HandleFunc("/api/v1/server-instances/{id}/files/browse", h.serverFilesBrowse)
|
||||
mux.HandleFunc("/api/v1/server-instances/{id}/files/refresh", h.serverFilesRefresh)
|
||||
mux.HandleFunc("/api/v1/server-instances/{id}/files/read-snapshot", h.serverFilesReadSnapshot)
|
||||
mux.HandleFunc("/api/v1/server-instances/{id}/files/read", h.serverFilesRead)
|
||||
@@ -1331,6 +1332,44 @@ func (h *coreHandlers) serverFilesList(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, dto.ServerFileListFromDomain(result))
|
||||
}
|
||||
|
||||
// serverFilesBrowse godoc
|
||||
// @Summary Browse one live server directory
|
||||
// @Description Dispatches a fresh Run files.list job for the requested logical directory and briefly waits for the matching result without exposing host paths or stale cached listings.
|
||||
// @Tags server-files
|
||||
// @Accept json
|
||||
// @Produce json
|
||||
// @Param id path string true "Server instance ID"
|
||||
// @Param body body dto.ServerFileListRequest true "Server file browse request"
|
||||
// @Success 202 {object} dto.ServerFileListResponse
|
||||
// @Failure 400 {object} dto.ErrorResponse
|
||||
// @Failure 401 {object} dto.ErrorResponse
|
||||
// @Failure 403 {object} dto.ErrorResponse
|
||||
// @Failure 404 {object} dto.ErrorResponse
|
||||
// @Failure 405 {object} dto.ErrorResponse
|
||||
// @Router /api/v1/server-instances/{id}/files/browse [post]
|
||||
func (h *coreHandlers) serverFilesBrowse(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeMethodNotAllowed(w, http.MethodPost)
|
||||
return
|
||||
}
|
||||
request, err := decodeJSON[dto.ServerFileListRequest](r)
|
||||
if err != nil {
|
||||
writeDecodeError(w, err)
|
||||
return
|
||||
}
|
||||
request.DirectoryKey, err = h.serverFileDirectoryKey(r, request.DirectoryKey)
|
||||
if err != nil {
|
||||
writeServiceError(w, err)
|
||||
return
|
||||
}
|
||||
result, err := h.core.BrowseServerFilesForSession(r.Context(), bearerToken(r), request.ToDomain(r.PathValue("id")))
|
||||
if err != nil {
|
||||
writeServiceError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusAccepted, dto.ServerFileListFromDomain(result))
|
||||
}
|
||||
|
||||
func (h *coreHandlers) serverFilesRefresh(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeMethodNotAllowed(w, http.MethodPost)
|
||||
@@ -1702,7 +1741,7 @@ func (h *coreHandlers) runJobClaim(w http.ResponseWriter, r *http.Request) {
|
||||
writeDecodeError(w, err)
|
||||
return
|
||||
}
|
||||
result, err := h.core.ClaimRunJob(request.ToDomain())
|
||||
result, err := h.core.ClaimRunJobWithWait(r.Context(), request.ToDomain())
|
||||
if err != nil {
|
||||
writeServiceError(w, err)
|
||||
return
|
||||
|
||||
@@ -17,7 +17,7 @@ Routes use JSON request and response bodies unless a route explicitly accepts fi
|
||||
| Server runtime distribution | n/a | `GET /api/v1/server-instances/{id}/runtime/actions`, `POST /api/v1/server-instances/{id}/run/generate`, `POST /api/v1/server-instances/{id}/run/download`, `POST /api/v1/server-instances/{id}/run/key/reset`, `POST /api/v1/server-instances/{id}/run/update`, `GET /api/v1/server-instances/{id}/run/update`, `POST /api/v1/server-instances/{id}/client-managers/generate`, `POST /api/v1/server-instances/{id}/client-managers/download`, `POST /api/v1/server-instances/{id}/client-managers/key/reset`, `GET /api/v1/server-instances/{id}/dependencies`, `POST /api/v1/server-instances/{id}/dependencies/check`, `POST /api/v1/server-instances/{id}/dependencies/install` | `ServerRuntimeActionsResponse`, `RunDistributionGenerateRequest`, `RunDistributionResponse`, `RunUpdateRequest`, `RunUpdateJobResponse`/`RunUpdateJobListResponse`, `ClientManagerBuildRequest`, `ClientManagerDistributionResponse`, `ClientManagerDownloadRequest`, `ComponentKeyResetRequest`, `ComponentKeyResponse`, `DependencyCatalogResponse`, `DependencyJobRequest` |
|
||||
| Metrics | `GET /api/v1/metrics/platform`, `GET /api/v1/metrics/server-instances` | n/a | `PlatformResourceUsageResponse`, `ServerMetricsResponse`, `ServerMetricsListResponse` |
|
||||
| File operations | `POST /api/v1/file-operations/dispatch` | n/a | `FileOperationDispatchRequest`, `FileOperationDispatchResponse` |
|
||||
| Server file manager | `GET /api/v1/server-instances/{id}/files/workspace`, `GET /api/v1/server-instances/{id}/files/list`, `POST /api/v1/server-instances/{id}/files/refresh`, `POST /api/v1/server-instances/{id}/files/read`, `POST /api/v1/server-instances/{id}/files/write`, `POST /api/v1/server-instances/{id}/files/upload`, `POST /api/v1/server-instances/{id}/files/download` | `GET /api/v1/server-instances/{id}/files/read-snapshot` | `ServerFileWorkspaceResponse`, `ServerFileListResponse`, `DeclaredFileReadSnapshotResponse`, `ServerFileReadRequest`, `ServerFileWriteRequest`, `ServerFileUploadResponse`, `ServerFileDownloadRequest`, `ServerFileDownloadResponse` |
|
||||
| Server file manager | `GET /api/v1/server-instances/{id}/files/workspace`, `POST /api/v1/server-instances/{id}/files/browse`, `GET /api/v1/server-instances/{id}/files/list`, `POST /api/v1/server-instances/{id}/files/refresh`, `POST /api/v1/server-instances/{id}/files/read`, `POST /api/v1/server-instances/{id}/files/write`, `POST /api/v1/server-instances/{id}/files/upload`, `POST /api/v1/server-instances/{id}/files/download` | `GET /api/v1/server-instances/{id}/files/read-snapshot` | `ServerFileWorkspaceResponse`, `ServerFileListResponse`, `DeclaredFileReadSnapshotResponse`, `ServerFileReadRequest`, `ServerFileWriteRequest`, `ServerFileUploadResponse`, `ServerFileDownloadRequest`, `ServerFileDownloadResponse` |
|
||||
| Plugin-owned data | n/a | `GET/PUT/DELETE /api/v1/server-instances/{id}/plugin-data/{collection}`, `POST .../plugin-data/{collection}/transaction` | `PluginDataPutRequest`, `PluginDataTransactionRequest`, `PluginDataRecordResponse`, `PluginDataListResponse` |
|
||||
| Server administrators | `GET /api/v1/server-instances/{id}/administrators/candidates`, `POST /api/v1/server-instances/{id}/administrators` | `DELETE /api/v1/server-instances/{id}/administrators/{userId}` | `ServerMemberRequest`, `ServerMemberResponse`, `ServerMemberListResponse`, `ServerInstanceResponse` |
|
||||
| Run endpoints | `GET /api/v1/run/endpoints`, `POST /api/v1/run/endpoints` | `GET /api/v1/run/endpoints/{id}` | `RunEndpointCreateRequest`, `RunEndpointResponse`, `RunEndpointListResponse` |
|
||||
@@ -176,7 +176,7 @@ Control is the highest-priority run-facing channel; artifact/file transfer press
|
||||
|
||||
## Implemented Run Job Actions
|
||||
|
||||
- `POST /api/v1/run/jobs/claim`: accept `RunJobClaimRequest`, validate the active Run session, sweep expired endpoint work, and durably claim one eligible queued/retrying job with a monotonic per-job attempt, hashed lease credential, ack deadline, and execution lease.
|
||||
- `POST /api/v1/run/jobs/claim`: accept `RunJobClaimRequest`, validate the active Run session, optionally hold the request for a bounded `waitSeconds` window, sweep expired endpoint work, and durably claim one eligible queued/retrying job with a monotonic per-job attempt, hashed lease credential, ack deadline, and execution lease.
|
||||
- `POST /api/v1/run/jobs/ack`: accept `RunJobAckRequest`, fence endpoint/session generation/attempt/lease, reject late acknowledgements, and move the current attempt into running state.
|
||||
- `POST /api/v1/run/jobs/progress`: accept `RunJobProgressRequest`, reject stale sequences and expired/old attempts, persist bounded progress, and renew the current execution lease.
|
||||
- `POST /api/v1/run/jobs/result`: accept `RunJobResultRequest` and write an idempotent terminal result or durable retry-wait transition with capped exponential backoff.
|
||||
|
||||
@@ -37,6 +37,7 @@ type RunJobClaim struct {
|
||||
SessionToken string
|
||||
Capabilities []string
|
||||
Capacity RunCapacity
|
||||
WaitSeconds int
|
||||
}
|
||||
|
||||
type RunJobClaimResult struct {
|
||||
|
||||
@@ -35,6 +35,7 @@ type RunJobClaimRequest struct {
|
||||
SessionToken string `json:"sessionToken"`
|
||||
Capabilities []string `json:"capabilities"`
|
||||
Capacity RunCapacityResponse `json:"capacity"`
|
||||
WaitSeconds int `json:"waitSeconds,omitempty"`
|
||||
}
|
||||
|
||||
type RunJobClaimResponse struct {
|
||||
@@ -405,6 +406,7 @@ func (request RunJobClaimRequest) ToDomain() domain.RunJobClaim {
|
||||
SessionToken: request.SessionToken,
|
||||
Capabilities: domain.CopyStringSlice(request.Capabilities),
|
||||
Capacity: capacityToDomain(request.Capacity),
|
||||
WaitSeconds: request.WaitSeconds,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
@@ -22,6 +23,8 @@ type ArtifactBodyStore interface {
|
||||
LoadTransfers() ([]domain.ArtifactTransferSession, error)
|
||||
PutPayload(string, []byte) error
|
||||
GetPayload(string) ([]byte, error)
|
||||
ReadPayloadRange(string, int64, int) ([]byte, error)
|
||||
CommitTransferPayload(domain.ArtifactTransferSession) error
|
||||
}
|
||||
|
||||
type MemoryArtifactBodyStore struct {
|
||||
@@ -73,6 +76,40 @@ func (store *MemoryArtifactBodyStore) GetPayload(artifactID string) ([]byte, err
|
||||
return domain.CopyBytes(payload), nil
|
||||
}
|
||||
|
||||
func (store *MemoryArtifactBodyStore) ReadPayloadRange(artifactID string, offset int64, length int) ([]byte, error) {
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
payload, exists := store.payloads[artifactID]
|
||||
if !exists {
|
||||
return nil, repo.ErrNotFound
|
||||
}
|
||||
if offset < 0 || offset > int64(len(payload)) || length < 0 || int64(length) > int64(len(payload))-offset {
|
||||
return nil, fmt.Errorf("artifact range is invalid")
|
||||
}
|
||||
return domain.CopyBytes(payload[int(offset) : int(offset)+length]), nil
|
||||
}
|
||||
|
||||
func (store *MemoryArtifactBodyStore) CommitTransferPayload(session domain.ArtifactTransferSession) error {
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
payload := make([]byte, 0, int(session.SizeBytes))
|
||||
for index := 0; index < session.TotalChunks; index++ {
|
||||
record, exists := session.ReceivedChunks[index]
|
||||
if !exists {
|
||||
return validationError("artifact transfer has missing chunks")
|
||||
}
|
||||
payload = append(payload, record.Payload...)
|
||||
}
|
||||
if int64(len(payload)) != session.SizeBytes {
|
||||
return validationError("artifact transfer size does not match metadata")
|
||||
}
|
||||
if checksum := validator.BytesChecksum(payload); checksum != session.Checksum {
|
||||
return validationError("artifact transfer checksum does not match metadata")
|
||||
}
|
||||
store.payloads[session.ArtifactID] = payload
|
||||
return nil
|
||||
}
|
||||
|
||||
type FileArtifactBodyStore struct {
|
||||
mu sync.Mutex
|
||||
rootDir string
|
||||
@@ -101,12 +138,14 @@ func (store *FileArtifactBodyStore) SaveTransfer(session domain.ArtifactTransfer
|
||||
}
|
||||
manifest := domain.CopyArtifactTransferSession(session)
|
||||
for index, record := range manifest.ReceivedChunks {
|
||||
payload := domain.CopyBytes(record.Payload)
|
||||
if len(payload) != record.SizeBytes || validator.BytesChecksum(payload) != record.Checksum {
|
||||
return validationError("artifact chunk does not match durable manifest")
|
||||
}
|
||||
if err := writeAtomicFile(filepath.Join(dir, fmt.Sprintf("chunk-%08d.bin", index)), payload, 0o600); err != nil {
|
||||
return err
|
||||
if record.Payload != nil {
|
||||
payload := domain.CopyBytes(record.Payload)
|
||||
if len(payload) != record.SizeBytes || validator.BytesChecksum(payload) != record.Checksum {
|
||||
return validationError("artifact chunk does not match durable manifest")
|
||||
}
|
||||
if err := writeAtomicFile(filepath.Join(dir, fmt.Sprintf("chunk-%08d.bin", index)), payload, 0o600); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
record.Payload = nil
|
||||
manifest.ReceivedChunks[index] = record
|
||||
@@ -148,14 +187,11 @@ func (store *FileArtifactBodyStore) LoadTransfers() ([]domain.ArtifactTransferSe
|
||||
return nil, fmt.Errorf("artifact transfer manifest identity mismatch")
|
||||
}
|
||||
for index, record := range session.ReceivedChunks {
|
||||
payload, err := os.ReadFile(filepath.Join(dir, fmt.Sprintf("chunk-%08d.bin", index)))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("read artifact transfer chunk: %w", err)
|
||||
chunkPath := filepath.Join(dir, fmt.Sprintf("chunk-%08d.bin", index))
|
||||
if _, err := os.Stat(chunkPath); err != nil {
|
||||
return nil, fmt.Errorf("stat artifact transfer chunk: %w", err)
|
||||
}
|
||||
if len(payload) != record.SizeBytes || validator.BytesChecksum(payload) != record.Checksum {
|
||||
return nil, validationError("durable artifact chunk checksum mismatch")
|
||||
}
|
||||
record.Payload = payload
|
||||
record.Payload = nil
|
||||
session.ReceivedChunks[index] = record
|
||||
}
|
||||
out = append(out, domain.CopyArtifactTransferSession(session))
|
||||
@@ -182,6 +218,107 @@ func (store *FileArtifactBodyStore) GetPayload(artifactID string) ([]byte, error
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
func (store *FileArtifactBodyStore) ReadPayloadRange(artifactID string, offset int64, length int) ([]byte, error) {
|
||||
if offset < 0 || length < 0 {
|
||||
return nil, fmt.Errorf("artifact range is invalid")
|
||||
}
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
file, err := os.Open(store.payloadPath(artifactID))
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil, repo.ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("open artifact payload: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
payload := make([]byte, length)
|
||||
read, err := file.ReadAt(payload, offset)
|
||||
if err != nil && !(errors.Is(err, io.ErrUnexpectedEOF) && read == length) {
|
||||
return nil, fmt.Errorf("read artifact payload range: %w", err)
|
||||
}
|
||||
if read != length {
|
||||
return nil, fmt.Errorf("artifact payload range is shorter than requested")
|
||||
}
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
func (store *FileArtifactBodyStore) CommitTransferPayload(session domain.ArtifactTransferSession) error {
|
||||
store.mu.Lock()
|
||||
defer store.mu.Unlock()
|
||||
|
||||
transferDir := store.transferDir(session.TransferID)
|
||||
payloadPath := store.payloadPath(session.ArtifactID)
|
||||
if err := os.MkdirAll(filepath.Dir(payloadPath), 0o700); err != nil {
|
||||
return fmt.Errorf("create artifact payload directory: %w", err)
|
||||
}
|
||||
tmp := payloadPath + ".tmp"
|
||||
out, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open artifact payload temporary file: %w", err)
|
||||
}
|
||||
hash := sha256.New()
|
||||
written := int64(0)
|
||||
for index := 0; index < session.TotalChunks; index++ {
|
||||
record, exists := session.ReceivedChunks[index]
|
||||
if !exists {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return validationError("artifact transfer has missing chunks")
|
||||
}
|
||||
chunkPath := filepath.Join(transferDir, fmt.Sprintf("chunk-%08d.bin", index))
|
||||
chunk, err := os.Open(chunkPath)
|
||||
if err != nil {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return fmt.Errorf("open artifact transfer chunk: %w", err)
|
||||
}
|
||||
chunkHash := sha256.New()
|
||||
count, copyErr := io.Copy(io.MultiWriter(out, hash, chunkHash), chunk)
|
||||
closeErr := chunk.Close()
|
||||
if copyErr != nil {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return fmt.Errorf("copy artifact transfer chunk: %w", copyErr)
|
||||
}
|
||||
if closeErr != nil {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return fmt.Errorf("close artifact transfer chunk: %w", closeErr)
|
||||
}
|
||||
if int(count) != record.SizeBytes || "sha256:"+hex.EncodeToString(chunkHash.Sum(nil)) != record.Checksum {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return validationError("durable artifact chunk checksum mismatch")
|
||||
}
|
||||
written += count
|
||||
}
|
||||
if written != session.SizeBytes {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return validationError("artifact transfer size does not match metadata")
|
||||
}
|
||||
if checksum := "sha256:" + hex.EncodeToString(hash.Sum(nil)); checksum != session.Checksum {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return validationError("artifact transfer checksum does not match metadata")
|
||||
}
|
||||
if err := out.Sync(); err != nil {
|
||||
_ = out.Close()
|
||||
_ = os.Remove(tmp)
|
||||
return fmt.Errorf("sync artifact payload: %w", err)
|
||||
}
|
||||
if err := out.Close(); err != nil {
|
||||
_ = os.Remove(tmp)
|
||||
return fmt.Errorf("close artifact payload: %w", err)
|
||||
}
|
||||
if err := os.Rename(tmp, payloadPath); err != nil {
|
||||
_ = os.Remove(tmp)
|
||||
return fmt.Errorf("commit artifact payload: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (store *FileArtifactBodyStore) transferDir(transferID string) string {
|
||||
return filepath.Join(store.rootDir, "transfers", stableStorageKey(transferID))
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
)
|
||||
|
||||
const artifactDownloadStorageBehavior = "platform-durable-artifact-store"
|
||||
const artifactPayloadCacheLimit = 4 * 1024 * 1024
|
||||
|
||||
func (svc *CoreService) GetArtifactForSession(sessionID string, artifactID string) (domain.Artifact, error) {
|
||||
artifact, err := svc.store.Artifacts().Get(strings.TrimSpace(artifactID))
|
||||
@@ -73,29 +74,22 @@ func (svc *CoreService) ReadArtifactContentForSession(sessionID string, request
|
||||
if artifact.State != domain.ArtifactStateAvailable {
|
||||
return domain.ArtifactContent{}, validationError("artifact must be available before download")
|
||||
}
|
||||
payload, err := svc.artifactPayload(artifact.ID)
|
||||
if err != nil {
|
||||
return domain.ArtifactContent{}, err
|
||||
}
|
||||
if int64(len(payload)) != artifact.SizeBytes {
|
||||
return domain.ArtifactContent{}, validationError("artifact content size does not match metadata")
|
||||
}
|
||||
if checksum := validator.BytesChecksum(payload); checksum != artifact.Checksum {
|
||||
return domain.ArtifactContent{}, validationError("artifact content checksum does not match metadata")
|
||||
}
|
||||
if request.Offset >= artifact.SizeBytes {
|
||||
return domain.ArtifactContent{}, validationError("offset must be inside artifact content")
|
||||
}
|
||||
limit := request.Limit
|
||||
if limit == 0 {
|
||||
limit = validator.MaxArtifactDownloadBytes
|
||||
}
|
||||
if request.Offset > artifact.SizeBytes {
|
||||
return domain.ArtifactContent{}, validationError("artifact range exceeds metadata")
|
||||
}
|
||||
remaining := artifact.SizeBytes - request.Offset
|
||||
if int64(limit) > remaining {
|
||||
limit = int(remaining)
|
||||
}
|
||||
end := int(request.Offset) + limit
|
||||
part := domain.CopyBytes(payload[int(request.Offset):end])
|
||||
payload, err := svc.artifactStore.ReadPayloadRange(artifact.ID, request.Offset, limit)
|
||||
if err != nil {
|
||||
return domain.ArtifactContent{}, err
|
||||
}
|
||||
part := domain.CopyBytes(payload)
|
||||
filename, contentType := svc.artifactDownloadPresentation(artifact)
|
||||
content := domain.ArtifactContent{
|
||||
ArtifactID: artifact.ID,
|
||||
@@ -162,7 +156,7 @@ func (svc *CoreService) artifactPayload(artifactID string) ([]byte, error) {
|
||||
return domain.CopyBytes(payload), nil
|
||||
}
|
||||
if payload, err := svc.artifactStore.GetPayload(artifactID); err == nil {
|
||||
svc.artifactPayloads[artifactID] = domain.CopyBytes(payload)
|
||||
svc.cacheArtifactPayload(artifactID, payload)
|
||||
return payload, nil
|
||||
} else if !errors.Is(err, repo.ErrNotFound) {
|
||||
return nil, err
|
||||
@@ -193,10 +187,18 @@ func (svc *CoreService) artifactPayload(artifactID string) ([]byte, error) {
|
||||
if err := svc.artifactStore.PutPayload(artifactID, payload); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
svc.artifactPayloads[artifactID] = domain.CopyBytes(payload)
|
||||
svc.cacheArtifactPayload(artifactID, payload)
|
||||
return payload, nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) cacheArtifactPayload(artifactID string, payload []byte) {
|
||||
if len(payload) > artifactPayloadCacheLimit {
|
||||
delete(svc.artifactPayloads, artifactID)
|
||||
return
|
||||
}
|
||||
svc.artifactPayloads[artifactID] = domain.CopyBytes(payload)
|
||||
}
|
||||
|
||||
func artifactDownloadFilename(artifactID string) string {
|
||||
name := strings.TrimSpace(artifactID)
|
||||
if name == "" || strings.Contains(name, "/") || strings.Contains(name, `\`) || strings.Contains(name, "://") {
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
@@ -111,7 +110,7 @@ func (svc *CoreService) UploadArtifactChunk(chunk domain.ArtifactChunkUpload) (d
|
||||
}
|
||||
|
||||
if existing, exists := session.ReceivedChunks[chunk.ChunkIndex]; exists {
|
||||
if existing.Offset == chunk.Offset && existing.SizeBytes == chunk.SizeBytes && existing.Checksum == chunk.Checksum && bytes.Equal(existing.Payload, chunk.Payload) {
|
||||
if existing.Offset == chunk.Offset && existing.SizeBytes == chunk.SizeBytes && existing.Checksum == chunk.Checksum {
|
||||
return artifactChunkUploadResult(session, chunk.ChunkIndex, true, stamp), nil
|
||||
}
|
||||
return domain.ArtifactChunkUploadResult{}, validationError("artifact chunk conflicts with acknowledged chunk")
|
||||
@@ -129,7 +128,7 @@ func (svc *CoreService) UploadArtifactChunk(chunk domain.ArtifactChunkUpload) (d
|
||||
if err := svc.artifactStore.SaveTransfer(session); err != nil {
|
||||
return domain.ArtifactChunkUploadResult{}, err
|
||||
}
|
||||
svc.artifactTransfers[session.TransferID] = domain.CopyArtifactTransferSession(session)
|
||||
svc.artifactTransfers[session.TransferID] = svc.artifactTransferSessionForMemory(session)
|
||||
return artifactChunkUploadResult(session, chunk.ChunkIndex, false, stamp), nil
|
||||
}
|
||||
|
||||
@@ -185,19 +184,13 @@ func (svc *CoreService) CompleteArtifactTransfer(complete domain.ArtifactTransfe
|
||||
return domain.ArtifactTransferCompleteResult{}, validationError("artifact transfer has missing chunks")
|
||||
}
|
||||
|
||||
payload := make([]byte, 0, int(session.SizeBytes))
|
||||
for index := 0; index < session.TotalChunks; index++ {
|
||||
record, exists := session.ReceivedChunks[index]
|
||||
if !exists {
|
||||
if _, exists := session.ReceivedChunks[index]; !exists {
|
||||
return domain.ArtifactTransferCompleteResult{}, validationError("artifact transfer has missing chunks")
|
||||
}
|
||||
payload = append(payload, record.Payload...)
|
||||
}
|
||||
if int64(len(payload)) != session.SizeBytes {
|
||||
return domain.ArtifactTransferCompleteResult{}, validationError("artifact transfer size does not match metadata")
|
||||
}
|
||||
if checksum := validator.BytesChecksum(payload); checksum != session.Checksum {
|
||||
return domain.ArtifactTransferCompleteResult{}, validationError("artifact transfer checksum does not match metadata")
|
||||
if err := svc.artifactStore.CommitTransferPayload(session); err != nil {
|
||||
return domain.ArtifactTransferCompleteResult{}, err
|
||||
}
|
||||
|
||||
artifact.SizeBytes = session.SizeBytes
|
||||
@@ -207,22 +200,29 @@ func (svc *CoreService) CompleteArtifactTransfer(complete domain.ArtifactTransfe
|
||||
if err := validator.ValidateArtifact(artifact); err != nil {
|
||||
return domain.ArtifactTransferCompleteResult{}, err
|
||||
}
|
||||
if err := svc.artifactStore.PutPayload(artifact.ID, payload); err != nil {
|
||||
return domain.ArtifactTransferCompleteResult{}, err
|
||||
}
|
||||
if err := svc.store.Artifacts().Update(artifact); err != nil {
|
||||
return domain.ArtifactTransferCompleteResult{}, err
|
||||
}
|
||||
svc.artifactPayloads[artifact.ID] = domain.CopyBytes(payload)
|
||||
session.Completed = true
|
||||
session.UpdatedAt = stamp
|
||||
if err := svc.artifactStore.SaveTransfer(session); err != nil {
|
||||
return domain.ArtifactTransferCompleteResult{}, err
|
||||
}
|
||||
svc.artifactTransfers[session.TransferID] = domain.CopyArtifactTransferSession(session)
|
||||
svc.artifactTransfers[session.TransferID] = svc.artifactTransferSessionForMemory(session)
|
||||
return domain.ArtifactTransferCompleteResult{Accepted: true, TransferID: session.TransferID, Artifact: artifact, Completed: true, ServerTime: stamp}, nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) artifactTransferSessionForMemory(session domain.ArtifactTransferSession) domain.ArtifactTransferSession {
|
||||
cached := domain.CopyArtifactTransferSession(session)
|
||||
if _, durableFileStore := svc.artifactStore.(*FileArtifactBodyStore); durableFileStore {
|
||||
for index, record := range cached.ReceivedChunks {
|
||||
record.Payload = nil
|
||||
cached.ReceivedChunks[index] = record
|
||||
}
|
||||
}
|
||||
return cached
|
||||
}
|
||||
|
||||
func (svc *CoreService) getArtifactTransferSession(transferID string) (domain.ArtifactTransferSession, error) {
|
||||
session, exists := svc.artifactTransfers[transferID]
|
||||
if !exists {
|
||||
|
||||
@@ -433,7 +433,7 @@ func (svc *CoreService) storeDistributionBuildArtifact(artifactID string, jobID
|
||||
} else if err := svc.store.Artifacts().Update(artifact); err != nil {
|
||||
return err
|
||||
}
|
||||
svc.artifactPayloads[artifact.ID] = domain.CopyBytes(payload)
|
||||
svc.cacheArtifactPayload(artifact.ID, payload)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -932,7 +932,7 @@ func (svc *CoreService) ensureArtifactPayload(artifactID string, payload []byte,
|
||||
if int64(len(existingPayload)) != artifact.SizeBytes || validator.BytesChecksum(existingPayload) != artifact.Checksum {
|
||||
return validationError("artifact payload does not match metadata")
|
||||
}
|
||||
svc.artifactPayloads[artifactID] = domain.CopyBytes(existingPayload)
|
||||
svc.cacheArtifactPayload(artifactID, existingPayload)
|
||||
return nil
|
||||
} else if !errors.Is(err, repo.ErrNotFound) {
|
||||
return err
|
||||
@@ -943,7 +943,7 @@ func (svc *CoreService) ensureArtifactPayload(artifactID string, payload []byte,
|
||||
if err := svc.artifactStore.PutPayload(artifactID, payload); err != nil {
|
||||
return err
|
||||
}
|
||||
svc.artifactPayloads[artifactID] = domain.CopyBytes(payload)
|
||||
svc.cacheArtifactPayload(artifactID, payload)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -91,8 +90,8 @@ func TestFileArtifactBodyStoreResumesTransferAfterServiceRestart(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatalf("final restart service: %v", err)
|
||||
}
|
||||
stored, err := finalService.artifactPayload("durable-artifact")
|
||||
if err != nil || !bytes.Equal(stored, payload) {
|
||||
stored, err := finalService.artifactStore.ReadPayloadRange("durable-artifact", 0, len(payload))
|
||||
if err != nil || string(stored) != string(payload) {
|
||||
t.Fatalf("expected durable payload after restart, payload=%q err=%v", stored, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"fmt"
|
||||
"sort"
|
||||
@@ -85,6 +86,68 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai
|
||||
}), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) ClaimRunJobWithWait(ctx context.Context, claim domain.RunJobClaim) (domain.RunJobClaimResult, error) {
|
||||
claim = domain.CopyRunJobClaim(claim)
|
||||
result, err := svc.ClaimRunJob(claim)
|
||||
if err != nil || result.HasJob || claim.WaitSeconds <= 0 || runJobClaimAtCapacity(claim) {
|
||||
return result, err
|
||||
}
|
||||
waiter := svc.registerRunJobWaiter(claim.RunEndpointID)
|
||||
defer svc.unregisterRunJobWaiter(claim.RunEndpointID, waiter)
|
||||
result, err = svc.ClaimRunJob(claim)
|
||||
if err != nil || result.HasJob {
|
||||
return result, err
|
||||
}
|
||||
timer := time.NewTimer(time.Duration(claim.WaitSeconds) * time.Second)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return domain.RunJobClaimResult{}, ctx.Err()
|
||||
case <-waiter:
|
||||
case <-timer.C:
|
||||
}
|
||||
return svc.ClaimRunJob(claim)
|
||||
}
|
||||
|
||||
func runJobClaimAtCapacity(claim domain.RunJobClaim) bool {
|
||||
return claim.Capacity.MaxJobs > 0 && claim.Capacity.RunningJobs >= claim.Capacity.MaxJobs
|
||||
}
|
||||
|
||||
func (svc *CoreService) registerRunJobWaiter(runEndpointID string) chan struct{} {
|
||||
waiter := make(chan struct{})
|
||||
svc.jobWaitMu.Lock()
|
||||
svc.jobWaiters[runEndpointID] = append(svc.jobWaiters[runEndpointID], waiter)
|
||||
svc.jobWaitMu.Unlock()
|
||||
return waiter
|
||||
}
|
||||
|
||||
func (svc *CoreService) unregisterRunJobWaiter(runEndpointID string, waiter chan struct{}) {
|
||||
svc.jobWaitMu.Lock()
|
||||
waiters := svc.jobWaiters[runEndpointID]
|
||||
for index, candidate := range waiters {
|
||||
if candidate == waiter {
|
||||
waiters = append(waiters[:index], waiters[index+1:]...)
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(waiters) == 0 {
|
||||
delete(svc.jobWaiters, runEndpointID)
|
||||
} else {
|
||||
svc.jobWaiters[runEndpointID] = waiters
|
||||
}
|
||||
svc.jobWaitMu.Unlock()
|
||||
}
|
||||
|
||||
func (svc *CoreService) notifyRunJobWaiters(runEndpointID string) {
|
||||
svc.jobWaitMu.Lock()
|
||||
waiters := svc.jobWaiters[runEndpointID]
|
||||
delete(svc.jobWaiters, runEndpointID)
|
||||
svc.jobWaitMu.Unlock()
|
||||
for _, waiter := range waiters {
|
||||
close(waiter)
|
||||
}
|
||||
}
|
||||
|
||||
func withoutCapability(capabilities []string, forbidden string) []string {
|
||||
filtered := make([]string, 0, len(capabilities))
|
||||
for _, capability := range capabilities {
|
||||
@@ -570,7 +633,11 @@ func (svc *CoreService) updateScheduledJob(job domain.Job) error {
|
||||
if err := validator.ValidateJob(job); err != nil {
|
||||
return err
|
||||
}
|
||||
return svc.store.Jobs().Update(job)
|
||||
if err := svc.store.Jobs().Update(job); err != nil {
|
||||
return err
|
||||
}
|
||||
svc.notifyRunJobWaiters(job.RunEndpointID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func normalizeJobScheduling(job domain.Job, stamp time.Time) domain.Job {
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"browser.local/platform/domain"
|
||||
)
|
||||
@@ -98,6 +100,34 @@ func TestCoreServiceRunJobClaimNoJob(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoreServiceRunJobClaimWithWaitWakesOnCreate(t *testing.T) {
|
||||
svc, sessionToken := newRegisteredRunJobService(t)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
resultCh := make(chan domain.RunJobClaimResult, 1)
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
claim, err := svc.ClaimRunJobWithWait(ctx, domain.RunJobClaim{RunEndpointID: "run-local", SessionToken: sessionToken, Capabilities: []string{"process.start"}, Capacity: domain.RunCapacity{MaxJobs: 4}, WaitSeconds: 10})
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
return
|
||||
}
|
||||
resultCh <- claim
|
||||
}()
|
||||
waitForRunJobWaiter(t, svc, "run-local")
|
||||
createQueuedRunJob(t, svc, "job-wait", "idem-wait")
|
||||
select {
|
||||
case err := <-errCh:
|
||||
t.Fatalf("claim wait failed: %v", err)
|
||||
case claim := <-resultCh:
|
||||
if !claim.HasJob || claim.Job == nil || claim.Job.JobID != "job-wait" {
|
||||
t.Fatalf("expected wait claim to return created job, got %+v", claim)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatalf("claim wait timed out: %v", ctx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
func TestCoreServiceRunJobClaimSkipsServerFileCapabilityWithoutDeclaration(t *testing.T) {
|
||||
svc := newTestCoreService()
|
||||
plugin, endpoint := createPluginAndRunEndpoint(t, svc)
|
||||
@@ -314,3 +344,18 @@ func createQueuedRunJob(t *testing.T, svc *CoreService, id string, idempotencyKe
|
||||
}
|
||||
return job
|
||||
}
|
||||
|
||||
func waitForRunJobWaiter(t *testing.T, svc *CoreService, runEndpointID string) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(500 * time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
svc.jobWaitMu.Lock()
|
||||
count := len(svc.jobWaiters[runEndpointID])
|
||||
svc.jobWaitMu.Unlock()
|
||||
if count > 0 {
|
||||
return
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
t.Fatalf("waiter for %s was not registered", runEndpointID)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/pbkdf2"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
@@ -126,6 +127,7 @@ type Core interface {
|
||||
GetServerFileWorkspaceForSession(string, string) (domain.ServerFileWorkspaceView, error)
|
||||
ListServerFilesForSession(string, domain.ServerFileListRequest) (domain.ServerFileListResult, error)
|
||||
RefreshServerFileListForSession(string, domain.ServerFileListRequest) (domain.ServerFileListResult, error)
|
||||
BrowseServerFilesForSession(context.Context, string, domain.ServerFileListRequest) (domain.ServerFileListResult, error)
|
||||
ReadServerFileForSession(string, domain.ServerFileReadRequest) (domain.FileOperationDispatchResult, error)
|
||||
WriteServerFileForSession(string, domain.ServerFileWriteRequest) (domain.FileOperationDispatchResult, error)
|
||||
UploadServerFileForSession(string, domain.ServerFileUploadRequest) (domain.ServerFileUploadDispatch, error)
|
||||
@@ -140,6 +142,7 @@ type Core interface {
|
||||
ListJobsForSession(string, domain.JobFilter) ([]domain.Job, error)
|
||||
RequestRunJobCancelForSession(string, domain.RunJobCancelRequest) (domain.RunJobCancelRequestResult, error)
|
||||
ClaimRunJob(domain.RunJobClaim) (domain.RunJobClaimResult, error)
|
||||
ClaimRunJobWithWait(context.Context, domain.RunJobClaim) (domain.RunJobClaimResult, error)
|
||||
AckRunJob(domain.RunJobAck) (domain.RunJobAckResult, error)
|
||||
UpdateRunJobProgress(domain.RunJobProgress) (domain.RunJobProgressResult, error)
|
||||
CompleteRunJob(domain.RunJobResult) (domain.RunJobResultResult, error)
|
||||
@@ -231,6 +234,8 @@ type CoreService struct {
|
||||
runSessions map[string]domain.RunControlSession
|
||||
runSessionSeq uint64
|
||||
jobMu sync.Mutex
|
||||
jobWaitMu sync.Mutex
|
||||
jobWaiters map[string][]chan struct{}
|
||||
bridgeMu sync.Mutex
|
||||
bridgeSeq uint64
|
||||
logStore LogBodyStore
|
||||
@@ -279,6 +284,7 @@ func newCoreServiceWithLogStore(store repo.Store, logStore LogBodyStore, now fun
|
||||
now: now,
|
||||
authSessions: map[string]string{},
|
||||
runSessions: map[string]domain.RunControlSession{},
|
||||
jobWaiters: map[string][]chan struct{}{},
|
||||
logStore: logStore,
|
||||
logProjectionStates: map[string]map[string]pluginLogSequenceState{},
|
||||
logEventSubscribers: map[uint64]logEventSubscriber{},
|
||||
@@ -2516,6 +2522,9 @@ func (svc *CoreService) CreateJob(job domain.Job) (domain.Job, error) {
|
||||
if err := svc.ensureJobLogStreams(existing, stamp); err != nil {
|
||||
return domain.Job{}, err
|
||||
}
|
||||
if !isTerminalJobState(existing.State) {
|
||||
svc.notifyRunJobWaiters(existing.RunEndpointID)
|
||||
}
|
||||
return existing, nil
|
||||
}
|
||||
if !errors.Is(err, repo.ErrNotFound) {
|
||||
@@ -2555,6 +2564,7 @@ func (svc *CoreService) CreateJob(job domain.Job) (domain.Job, error) {
|
||||
if err := svc.ensureJobLogStreams(job, stamp); err != nil {
|
||||
return domain.Job{}, err
|
||||
}
|
||||
svc.notifyRunJobWaiters(job.RunEndpointID)
|
||||
return domain.CopyJob(job), nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -1004,6 +1006,70 @@ func TestServerFileListReportsFailedRuntimeRefresh(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerFileBrowseWaitsForFreshRunResultWithoutCachedList(t *testing.T) {
|
||||
svc := newTestCoreService()
|
||||
plugin, endpoint := createPluginAndRunEndpoint(t, svc)
|
||||
endpoint.Capabilities = append(endpoint.Capabilities, domain.JobCapabilityFilesList)
|
||||
if err := svc.store.RunEndpoints().Update(endpoint); err != nil {
|
||||
t.Fatalf("update file list capability: %v", err)
|
||||
}
|
||||
ownerSession := createServiceUserAndLogin(t, svc, domain.User{ID: "user-file-browse", DisplayName: "File Browse", Email: "file-browse@example.test", Roles: []string{"server-owner"}, PasswordHash: "secret-password"})
|
||||
instance, err := svc.CreateServerInstanceForSession(ownerSession, domain.ServerInstance{ID: "server-file-browse", PluginID: plugin.ID, RunEndpointID: endpoint.ID, Name: "File Browse Server", State: domain.ServerInstanceStateRunning, Deployment: domain.ServerDeploymentDefinition{Mode: domain.ServerDeploymentModeGuided, ServerRoot: `C:\scumserver`}})
|
||||
if err != nil {
|
||||
t.Fatalf("create server: %v", err)
|
||||
}
|
||||
createCompleteRuntimeBinding(t, svc, instance, "local")
|
||||
oldJob, err := svc.CreateJob(domain.Job{ID: "job-file-list-old", ServerInstanceID: instance.ID, RunEndpointID: endpoint.ID, Capability: domain.JobCapabilityFilesList, TargetKey: "server-root", IdempotencyKey: "idem-file-list-old"})
|
||||
if err != nil {
|
||||
t.Fatalf("create old list job: %v", err)
|
||||
}
|
||||
oldJob.State = domain.JobStateSucceeded
|
||||
oldJob.ExecutionResult = domain.JobExecutionResult{Kind: "file.list", Content: runFileListFixture("server-root", "", ".platform")}
|
||||
oldJob.TerminalAt = fixedTime.Add(10 * time.Minute)
|
||||
oldJob.UpdatedAt = oldJob.TerminalAt
|
||||
if err := svc.store.Jobs().Update(oldJob); err != nil {
|
||||
t.Fatalf("store old list result: %v", err)
|
||||
}
|
||||
helloRequest := validRunControlHello()
|
||||
helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, domain.JobCapabilityFilesList)
|
||||
helloRequest.CapabilityReport.Fingerprint = "cap-file-browse"
|
||||
hello, err := svc.RegisterRunHello(helloRequest)
|
||||
if err != nil {
|
||||
t.Fatalf("register run hello: %v", err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
resultCh := make(chan domain.ServerFileListResult, 1)
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
result, err := svc.BrowseServerFilesForSession(ctx, ownerSession, domain.ServerFileListRequest{ServerInstanceID: instance.ID, DirectoryKey: "server-root", IdempotencyKey: "idem-file-browse-fresh"})
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
return
|
||||
}
|
||||
resultCh <- result
|
||||
}()
|
||||
job := waitForServerFileListJob(t, svc, instance.ID, "job-file-list-old")
|
||||
claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, Capabilities: []string{domain.JobCapabilityFilesList}, Capacity: domain.RunCapacity{MaxJobs: 4}})
|
||||
if err != nil || !claim.HasJob || claim.Job.JobID != job.ID {
|
||||
t.Fatalf("claim fresh file list job: claim=%+v err=%v", claim, err)
|
||||
}
|
||||
_, err = svc.CompleteRunJob(domain.RunJobResult{RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100, Message: "listed"}, ExecutionResult: domain.JobExecutionResult{Kind: "file.list", Content: runFileListFixture("server-root", "", "SCUM")}})
|
||||
if err != nil {
|
||||
t.Fatalf("complete fresh file list job: %v", err)
|
||||
}
|
||||
select {
|
||||
case err := <-errCh:
|
||||
t.Fatalf("browse failed: %v", err)
|
||||
case result := <-resultCh:
|
||||
if result.State != "ready" || len(result.Entries) != 1 || result.Entries[0].Name != "SCUM" || result.Job.ID != job.ID {
|
||||
t.Fatalf("expected fresh browse result, got %+v", result)
|
||||
}
|
||||
case <-ctx.Done():
|
||||
t.Fatalf("browse timed out: %v", ctx.Err())
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerFileListFallsBackToPluginWorkspaceWithoutRunListCapability(t *testing.T) {
|
||||
svc := newTestCoreService()
|
||||
plugin, endpoint := createPluginAndRunEndpoint(t, svc)
|
||||
@@ -2062,6 +2128,29 @@ func createCompleteRuntimeBinding(t *testing.T, svc *CoreService, instance domai
|
||||
return binding
|
||||
}
|
||||
|
||||
func waitForServerFileListJob(t *testing.T, svc *CoreService, serverInstanceID string, excludedJobID string) domain.Job {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(500 * time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
jobs, err := svc.store.Jobs().List(domain.JobFilter{ServerInstanceID: serverInstanceID})
|
||||
if err != nil {
|
||||
t.Fatalf("list server jobs: %v", err)
|
||||
}
|
||||
for _, job := range jobs {
|
||||
if job.ID != excludedJobID && job.Capability == domain.JobCapabilityFilesList {
|
||||
return job
|
||||
}
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
t.Fatalf("fresh file list job was not created")
|
||||
return domain.Job{}
|
||||
}
|
||||
|
||||
func runFileListFixture(directoryKey string, relativePath string, name string) string {
|
||||
return fmt.Sprintf(`{"directoryKey":%q,"path":%q,"entries":[{"name":%q,"kind":"directory","directoryKey":%q,"relativePath":%q}]}`, directoryKey, relativePath, name, directoryKey, name)
|
||||
}
|
||||
|
||||
func runtimeBindingTestKeyIsSensitive(key string) bool {
|
||||
normalized := strings.ToLower(key)
|
||||
return strings.Contains(normalized, "password") || strings.Contains(normalized, "credential") || strings.Contains(normalized, "secret") || strings.Contains(normalized, "token") || strings.Contains(normalized, "dsn")
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -19,6 +20,7 @@ const (
|
||||
serverFileTransferChannel = "run-file-transfer"
|
||||
serverFileMaxInlineEditBytes = 64 * 1024
|
||||
serverFileDefaultDirectoryKey = "server-root"
|
||||
serverFileBrowseWait = 4 * time.Second
|
||||
)
|
||||
|
||||
type serverFileContext struct {
|
||||
@@ -188,6 +190,82 @@ func (svc *CoreService) RefreshServerFileListForSession(sessionID string, reques
|
||||
return domain.CopyServerFileListResult(domain.ServerFileListResult{ServerInstanceID: ctx.Instance.ID, PluginID: ctx.Plugin.ID, DirectoryKey: request.DirectoryKey, Path: request.Path, State: "pending", Entries: entries, Job: job, Reason: "目录刷新任务已派发到 Run。"}), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) BrowseServerFilesForSession(ctx context.Context, sessionID string, request domain.ServerFileListRequest) (domain.ServerFileListResult, error) {
|
||||
request = normalizeServerFileListRequest(request)
|
||||
if request.IdempotencyKey == "" {
|
||||
idempotencyKey, err := svc.serverFileBrowseIdempotencyKey(request)
|
||||
if err != nil {
|
||||
return domain.ServerFileListResult{}, err
|
||||
}
|
||||
request.IdempotencyKey = idempotencyKey
|
||||
}
|
||||
result, err := svc.RefreshServerFileListForSession(sessionID, request)
|
||||
if err != nil || result.State != "pending" || result.Job.ID == "" {
|
||||
return result, err
|
||||
}
|
||||
timer := time.NewTimer(serverFileBrowseWait)
|
||||
defer timer.Stop()
|
||||
for {
|
||||
current, ready, err := svc.serverFileListResultFromJob(request, result, result.Job.ID)
|
||||
if err != nil || ready {
|
||||
return current, err
|
||||
}
|
||||
waiter := svc.registerRunJobWaiter(result.Job.RunEndpointID)
|
||||
current, ready, err = svc.serverFileListResultFromJob(request, result, result.Job.ID)
|
||||
if err != nil || ready {
|
||||
svc.unregisterRunJobWaiter(result.Job.RunEndpointID, waiter)
|
||||
return current, err
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
svc.unregisterRunJobWaiter(result.Job.RunEndpointID, waiter)
|
||||
return domain.ServerFileListResult{}, ctx.Err()
|
||||
case <-waiter:
|
||||
case <-timer.C:
|
||||
svc.unregisterRunJobWaiter(result.Job.RunEndpointID, waiter)
|
||||
current.Reason = "正在读取目录。"
|
||||
return current, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (svc *CoreService) serverFileListResultFromJob(request domain.ServerFileListRequest, fallback domain.ServerFileListResult, jobID string) (domain.ServerFileListResult, bool, error) {
|
||||
job, err := svc.store.Jobs().Get(jobID)
|
||||
if err != nil {
|
||||
return domain.ServerFileListResult{}, true, err
|
||||
}
|
||||
result := domain.CopyServerFileListResult(fallback)
|
||||
result.Job = job
|
||||
if !isTerminalJobState(job.State) {
|
||||
return result, false, nil
|
||||
}
|
||||
result.RefreshedAt = job.TerminalAt
|
||||
if job.State != domain.JobStateSucceeded || job.ExecutionResult.Kind != "file.list" {
|
||||
result.State = "failed"
|
||||
result.Reason = serverFileListJobFailureReason(job)
|
||||
return result, true, nil
|
||||
}
|
||||
entries, err := serverFileEntriesFromRunList(job.ExecutionResult.Content, request.DirectoryKey, request.Path)
|
||||
if err != nil {
|
||||
result.State = "failed"
|
||||
result.Reason = "Run 返回的文件列表无法解析。"
|
||||
return result, true, nil
|
||||
}
|
||||
result.State = "ready"
|
||||
result.Entries = filterServerFileEntries(entries, request.Query)
|
||||
result.Reason = "目录读取完成。"
|
||||
return result, true, nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) serverFileBrowseIdempotencyKey(request domain.ServerFileListRequest) (string, error) {
|
||||
token, err := randomToken()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
raw := strings.Join([]string{request.ServerInstanceID, request.DirectoryKey, request.Path, request.Query, strconv.FormatBool(request.Recursive), token}, ":")
|
||||
return fmt.Sprintf("file-browse:%d", stableStringNumber(raw)), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) ReadServerFileForSession(sessionID string, request domain.ServerFileReadRequest) (domain.FileOperationDispatchResult, error) {
|
||||
if err := validator.ValidateServerFileReadRequest(request); err != nil {
|
||||
return domain.FileOperationDispatchResult{}, err
|
||||
@@ -545,7 +623,7 @@ func (svc *CoreService) putBrowserFileArtifact(artifact domain.Artifact, payload
|
||||
if err := svc.artifactStore.PutPayload(artifact.ID, payload); err != nil {
|
||||
return err
|
||||
}
|
||||
svc.artifactPayloads[artifact.ID] = domain.CopyBytes(payload)
|
||||
svc.cacheArtifactPayload(artifact.ID, payload)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
)
|
||||
|
||||
const maxJobChannelMessageLength = 256
|
||||
const maxRunJobClaimWaitSeconds = 30
|
||||
|
||||
func ValidateRunJobClaim(claim domain.RunJobClaim) error {
|
||||
var violations []string
|
||||
@@ -19,6 +20,9 @@ func ValidateRunJobClaim(claim domain.RunJobClaim) error {
|
||||
violations = append(violations, fmt.Sprintf("capabilities[%d] is required", i))
|
||||
}
|
||||
}
|
||||
if claim.WaitSeconds < 0 || claim.WaitSeconds > maxRunJobClaimWaitSeconds {
|
||||
violations = append(violations, fmt.Sprintf("waitSeconds must be between 0 and %d", maxRunJobClaimWaitSeconds))
|
||||
}
|
||||
return finish(violations)
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user