iop/apps/node/internal/workspace/runtime.go

// Package workspace owns the Node-private, request-scoped workspace catalog.
package workspace

import (
	"context"
	"errors"
	"io/fs"
	"os"
	"path/filepath"
	"runtime"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	"go.uber.org/zap"

	iop "iop/proto/gen/iop"
)

var (
	ErrClosed           = errors.New("workspace runtime is closed")
	ErrInvalidRequest   = errors.New("workspace request is invalid")
	ErrUnknownWorkspace = errors.New("workspace is not configured")
	ErrRequestConflict  = errors.New("workspace request binding conflicts")
)

const (
	completedCleanupLimit   = 256
	defaultCleanupTimeout   = 5 * time.Second
	maxInternalArtifactSize = 1 << 20
	maxCleanupArtifacts     = 4096
)

// Runtime keeps the authorities admitted from the Edge configuration. It does
// not retain a path that is re-resolved for an operation: every catalog entry
// owns an os.Root opened during validation.
type Runtime struct {
	mu                sync.RWMutex
	lifetime          sync.RWMutex
	closed            bool
	catalog           map[string]*catalogEntry
	requests          map[string]*Request
	commandsMu        sync.Mutex
	activeCommands    map[commandKey]*commandExecution
	cancelledCommands map[commandKey]struct{}
	cleanupMu         sync.Mutex
	cleanupCalls      map[string]*cleanupCall
	cleanupOrder      []string
	observer          *workspaceSafeObserver
}

type catalogEntry struct {
	ref               string
	root              *os.Root
	directory         *os.File
	device            uint64
	inode             uint64
	operations        map[iop.WorkspaceOperation]struct{}
	commands          map[string]commandTemplate
	environment       map[string]struct{}
	maxRead           int64
	maxWrite          int64
	maxOutput         int64
	maxCommandTimeout int64
	// beforeRename is a deterministic package-test seam for failures and
	// parent replacement after the temporary file is durable.
	beforeRename func() error
}

// RequestAuthority is the complete immutable authority admitted by Edge for a
// single coordinator request. Runtime.Open validates it against the selected
// catalog entry and retains a defensive copy.
type RequestAuthority struct {
	RequestID           string
	WorkspaceRef        string
	Operations          []iop.WorkspaceOperation
	CommandIDs          []string
	MaxReadBytes        int64
	MaxWriteBytes       int64
	MaxOutputBytes      int64
	MaxCommandTimeoutMS int64
}

// Request is a read-only binding between the immutable coordinator request id
// and one catalog entry. The derived internal prefix is deliberately not
// caller-provided.
type Request struct {
	mu                sync.Mutex
	id                string
	workspaceRef      string
	entry             *catalogEntry
	internalPrefix    string
	operations        map[iop.WorkspaceOperation]struct{}
	commandIDs        []string
	maxRead           int64
	maxWrite          int64
	maxOutput         int64
	maxCommandTimeout int64
	cleaning          bool
	artifacts         map[string]ownedArtifact
	ownedParents      []ownedArtifact
	correlation       string
}

type ownedArtifactKind uint8

const (
	ownedArtifactFile ownedArtifactKind = iota + 1
	ownedArtifactDirectory
)

type ownedArtifact struct {
	relative string
	kind     ownedArtifactKind
	device   uint64
	inode    uint64
}

type cleanupCall struct {
	done   chan struct{}
	result CleanupResult
}

// CleanupResult is a content-free terminal for one immutable request cleanup.
// Every concurrent or duplicate caller observes the same cached value.
type CleanupResult struct {
	Status           iop.WorkspaceStatus
	Code             iop.WorkspaceErrorCode
	CleanedProcesses int32
	CleanedArtifacts int32
}

// NewRuntime validates and opens the Node-private catalog. Empty catalogs are
// supported for mixed-version Nodes; a non-empty catalog is Mac-only.
func NewRuntime(configs []*iop.WorkspaceConfig, hostOS string, logger *zap.Logger) (*Runtime, error) {
	rt := &Runtime{
		catalog:           make(map[string]*catalogEntry, len(configs)),
		requests:          make(map[string]*Request),
		activeCommands:    make(map[commandKey]*commandExecution),
		cancelledCommands: make(map[commandKey]struct{}),
		cleanupCalls:      make(map[string]*cleanupCall),
		observer:          &workspaceSafeObserver{inner: newZapWorkspaceObserver(logger)},
	}
	if len(configs) == 0 {
		return rt, nil
	}
	if hostOS == "" {
		hostOS = runtime.GOOS
	}
	if hostOS != "darwin" {
		return nil, errors.New("workspace catalog requires darwin")
	}
	for _, cfg := range configs {
		entry, err := openCatalogEntry(cfg)
		if err != nil {
			_ = rt.Close()
			return nil, err
		}
		if _, duplicate := rt.catalog[entry.ref]; duplicate {
			_ = entry.root.Close()
			_ = entry.directory.Close()
			_ = rt.Close()
			return nil, errors.New("duplicate workspace ref")
		}
		rt.catalog[entry.ref] = entry
	}
	return rt, nil
}

func openCatalogEntry(cfg *iop.WorkspaceConfig) (*catalogEntry, error) {
	if cfg == nil || strings.TrimSpace(cfg.GetRef()) == "" || cfg.GetRef() != strings.TrimSpace(cfg.GetRef()) {
		return nil, errors.New("invalid workspace ref")
	}
	if cfg.GetPlatform() != "darwin" || cfg.GetRoot() == "" || !filepath.IsAbs(cfg.GetRoot()) || cfg.GetRoot() == "/" || filepath.Clean(cfg.GetRoot()) != cfg.GetRoot() {
		return nil, errors.New("invalid workspace root")
	}
	info, err := os.Lstat(cfg.GetRoot())
	if err != nil || info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
		return nil, errors.New("invalid workspace root")
	}
	device, inode, ok := fileIdentity(info)
	if !ok {
		return nil, errors.New("workspace root identity unavailable")
	}
	directory, err := os.Open(cfg.GetRoot())
	if err != nil {
		return nil, errors.New("workspace directory unavailable")
	}
	openedInfo, err := directory.Stat()
	if err != nil {
		_ = directory.Close()
		return nil, errors.New("workspace root changed while opening")
	}
	openedDevice, openedInode, openedOK := fileIdentity(openedInfo)
	if !openedOK || openedDevice != device || openedInode != inode || !openedInfo.IsDir() {
		_ = directory.Close()
		return nil, errors.New("workspace root changed while opening")
	}
	root, err := os.OpenRoot(cfg.GetRoot())
	if err != nil {
		_ = directory.Close()
		return nil, errors.New("workspace root unavailable")
	}
	rootInfo, err := root.Stat(".")
	if err != nil {
		_ = root.Close()
		_ = directory.Close()
		return nil, errors.New("workspace root changed while opening")
	}
	rootDevice, rootInode, rootOK := fileIdentity(rootInfo)
	if !rootOK || rootDevice != device || rootInode != inode || !rootInfo.IsDir() {
		_ = root.Close()
		_ = directory.Close()
		return nil, errors.New("workspace root changed while opening")
	}
	operations := make(map[iop.WorkspaceOperation]struct{}, len(cfg.GetOperations()))
	for _, operation := range cfg.GetOperations() {
		switch operation {
		case iop.WorkspaceOperation_WORKSPACE_OPERATION_READ,
			iop.WorkspaceOperation_WORKSPACE_OPERATION_LIST,
			iop.WorkspaceOperation_WORKSPACE_OPERATION_WRITE,
			iop.WorkspaceOperation_WORKSPACE_OPERATION_DELETE,
			iop.WorkspaceOperation_WORKSPACE_OPERATION_COMMAND:
			if _, duplicate := operations[operation]; duplicate {
				_ = root.Close()
				_ = directory.Close()
				return nil, errors.New("duplicate workspace operation")
			}
			operations[operation] = struct{}{}
		default:
			_ = root.Close()
			_ = directory.Close()
			return nil, errors.New("invalid workspace operation")
		}
	}
	commands := make(map[string]commandTemplate, len(cfg.GetCommands()))
	for _, command := range cfg.GetCommands() {
		if command == nil || strings.TrimSpace(command.GetId()) == "" || command.GetId() != strings.TrimSpace(command.GetId()) {
			_ = root.Close()
			_ = directory.Close()
			return nil, errors.New("invalid workspace command")
		}
		if _, duplicate := commands[command.GetId()]; duplicate {
			_ = root.Close()
			_ = directory.Close()
			return nil, errors.New("duplicate workspace command")
		}
		if !filepath.IsAbs(command.GetExecutable()) || filepath.Clean(command.GetExecutable()) != command.GetExecutable() || strings.IndexByte(command.GetExecutable(), 0) >= 0 {
			_ = root.Close()
			_ = directory.Close()
			return nil, errors.New("invalid workspace command")
		}
		args := append([]string(nil), command.GetArgs()...)
		payloadBytes := len(command.GetExecutable())
		if payloadBytes > commandLaunchPayloadLimit {
			_ = root.Close()
			_ = directory.Close()
			return nil, errors.New("invalid workspace command")
		}
		for _, arg := range args {
			if strings.IndexByte(arg, 0) >= 0 {
				_ = root.Close()
				_ = directory.Close()
				return nil, errors.New("invalid workspace command")
			}
			payloadBytes += len(arg)
			if payloadBytes > commandLaunchPayloadLimit {
				_ = root.Close()
				_ = directory.Close()
				return nil, errors.New("invalid workspace command")
			}
		}
		commands[command.GetId()] = commandTemplate{executable: command.GetExecutable(), args: args}
	}
	environment := make(map[string]struct{}, len(cfg.GetEnvironmentAllowlist()))
	for _, name := range cfg.GetEnvironmentAllowlist() {
		if !validEnvironmentName(name) || name == commandShimEnvironment {
			_ = root.Close()
			_ = directory.Close()
			return nil, errors.New("invalid workspace environment allowlist")
		}
		if _, duplicate := environment[name]; duplicate {
			_ = root.Close()
			_ = directory.Close()
			return nil, errors.New("duplicate workspace environment")
		}
		environment[name] = struct{}{}
	}
	_, readEnabled := operations[iop.WorkspaceOperation_WORKSPACE_OPERATION_READ]
	_, listEnabled := operations[iop.WorkspaceOperation_WORKSPACE_OPERATION_LIST]
	_, writeEnabled := operations[iop.WorkspaceOperation_WORKSPACE_OPERATION_WRITE]
	_, commandEnabled := operations[iop.WorkspaceOperation_WORKSPACE_OPERATION_COMMAND]
	invalidLimits := readEnabled && cfg.GetMaxReadBytes() <= 0 ||
		writeEnabled && cfg.GetMaxWriteBytes() <= 0 ||
		(listEnabled || commandEnabled) && cfg.GetMaxOutputBytes() <= 0 ||
		commandEnabled && (cfg.GetMaxCommandTimeoutMs() <= 0 || len(commands) == 0) ||
		!commandEnabled && len(commands) != 0
	if len(operations) == 0 || invalidLimits {
		_ = root.Close()
		_ = directory.Close()
		return nil, errors.New("invalid workspace limits")
	}
	return &catalogEntry{
		ref: cfg.GetRef(), root: root, directory: directory, device: device, inode: inode, operations: operations, commands: commands, environment: environment,
		maxRead: cfg.GetMaxReadBytes(), maxWrite: cfg.GetMaxWriteBytes(), maxOutput: cfg.GetMaxOutputBytes(), maxCommandTimeout: cfg.GetMaxCommandTimeoutMs(),
	}, nil
}

// Open freezes a request's catalog authority. A duplicate request is allowed
// only when it repeats the exact same immutable binding.
func (r *Runtime) Open(authority RequestAuthority) (*Request, error) {
	if !validRequestID(authority.RequestID) || strings.TrimSpace(authority.WorkspaceRef) == "" || authority.WorkspaceRef != strings.TrimSpace(authority.WorkspaceRef) {
		return nil, ErrInvalidRequest
	}
	r.cleanupMu.Lock()
	cleanupKnown := r.cleanupCalls[authority.RequestID] != nil
	r.cleanupMu.Unlock()
	if cleanupKnown {
		return nil, ErrRequestConflict
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.closed {
		return nil, ErrClosed
	}
	entry := r.catalog[authority.WorkspaceRef]
	if entry == nil {
		return nil, ErrUnknownWorkspace
	}
	normalized, err := normalizeAuthority(authority, entry)
	if err != nil {
		return nil, err
	}
	if existing := r.requests[normalized.RequestID]; existing != nil {
		if existing.matches(normalized) {
			return existing, nil
		}
		return nil, ErrRequestConflict
	}
	operations := make(map[iop.WorkspaceOperation]struct{}, len(normalized.Operations))
	for _, operation := range normalized.Operations {
		operations[operation] = struct{}{}
	}
	artifacts, ownedParents, err := initializeRequestArtifacts(entry, normalized.RequestID)
	if err != nil {
		return nil, ErrInvalidRequest
	}
	req := &Request{
		id: normalized.RequestID, workspaceRef: normalized.WorkspaceRef, entry: entry,
		internalPrefix: ".iop/job/" + normalized.RequestID,
		operations:     operations, commandIDs: append([]string(nil), normalized.CommandIDs...),
		maxRead: normalized.MaxReadBytes, maxWrite: normalized.MaxWriteBytes,
		maxOutput: normalized.MaxOutputBytes, maxCommandTimeout: normalized.MaxCommandTimeoutMS,
		artifacts: artifacts, ownedParents: ownedParents,
		correlation: newWorkspaceCorrelation(),
	}
	r.requests[normalized.RequestID] = req
	return req, nil
}

func normalizeAuthority(authority RequestAuthority, entry *catalogEntry) (RequestAuthority, error) {
	normalized := authority
	normalized.Operations = append([]iop.WorkspaceOperation(nil), authority.Operations...)
	sort.Slice(normalized.Operations, func(i, j int) bool { return normalized.Operations[i] < normalized.Operations[j] })
	normalized.CommandIDs = append([]string(nil), authority.CommandIDs...)
	sort.Strings(normalized.CommandIDs)
	if len(normalized.Operations) == 0 {
		return RequestAuthority{}, ErrInvalidRequest
	}
	requested := make(map[iop.WorkspaceOperation]struct{}, len(normalized.Operations))
	for _, operation := range normalized.Operations {
		if operation == iop.WorkspaceOperation_WORKSPACE_OPERATION_UNSPECIFIED {
			return RequestAuthority{}, ErrInvalidRequest
		}
		if _, allowed := entry.operations[operation]; !allowed {
			return RequestAuthority{}, ErrInvalidRequest
		}
		if _, duplicate := requested[operation]; duplicate {
			return RequestAuthority{}, ErrInvalidRequest
		}
		requested[operation] = struct{}{}
	}
	for index, commandID := range normalized.CommandIDs {
		if strings.TrimSpace(commandID) == "" || commandID != strings.TrimSpace(commandID) || index > 0 && normalized.CommandIDs[index-1] == commandID {
			return RequestAuthority{}, ErrInvalidRequest
		}
		if _, allowed := entry.commands[commandID]; !allowed {
			return RequestAuthority{}, ErrInvalidRequest
		}
	}
	_, readEnabled := requested[iop.WorkspaceOperation_WORKSPACE_OPERATION_READ]
	_, listEnabled := requested[iop.WorkspaceOperation_WORKSPACE_OPERATION_LIST]
	_, writeEnabled := requested[iop.WorkspaceOperation_WORKSPACE_OPERATION_WRITE]
	_, commandEnabled := requested[iop.WorkspaceOperation_WORKSPACE_OPERATION_COMMAND]
	if !validAuthorityLimit(readEnabled, normalized.MaxReadBytes, entry.maxRead) ||
		!validAuthorityLimit(writeEnabled, normalized.MaxWriteBytes, entry.maxWrite) ||
		!validAuthorityLimit(listEnabled || commandEnabled, normalized.MaxOutputBytes, entry.maxOutput) ||
		!validAuthorityLimit(commandEnabled, normalized.MaxCommandTimeoutMS, entry.maxCommandTimeout) ||
		commandEnabled != (len(normalized.CommandIDs) > 0) {
		return RequestAuthority{}, ErrInvalidRequest
	}
	return normalized, nil
}

func validAuthorityLimit(enabled bool, value, maximum int64) bool {
	if !enabled {
		return value == 0
	}
	return value > 0 && value <= maximum
}

func (r *Request) matches(authority RequestAuthority) bool {
	if r == nil || r.id != authority.RequestID || r.workspaceRef != authority.WorkspaceRef ||
		r.maxRead != authority.MaxReadBytes || r.maxWrite != authority.MaxWriteBytes ||
		r.maxOutput != authority.MaxOutputBytes || r.maxCommandTimeout != authority.MaxCommandTimeoutMS ||
		!slices.Equal(r.commandIDs, authority.CommandIDs) || len(r.operations) != len(authority.Operations) {
		return false
	}
	for _, operation := range authority.Operations {
		if _, ok := r.operations[operation]; !ok {
			return false
		}
	}
	return true
}

// Request returns the immutable binding only while the request is open.
func (r *Runtime) Request(requestID string) (*Request, error) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	if r.closed {
		return nil, ErrClosed
	}
	req := r.requests[requestID]
	if req == nil {
		return nil, ErrInvalidRequest
	}
	return req, nil
}

// CloseRequest removes a request authority. It is idempotent so lifecycle
// teardown can safely race duplicate terminal signals.
func (r *Runtime) CloseRequest(requestID string) {
	ctx, cancel := context.WithTimeout(context.Background(), defaultCleanupTimeout)
	defer cancel()
	_ = r.Cleanup(ctx, requestID)
}

// Close releases all admitted root handles. Active operations take a shared
// lifetime lock, so a root cannot be closed under an operation.
func (r *Runtime) Close() error {
	r.mu.Lock()
	if r.closed {
		r.mu.Unlock()
		return nil
	}
	r.closed = true
	requestIDs := make([]string, 0, len(r.requests))
	for requestID := range r.requests {
		requestIDs = append(requestIDs, requestID)
	}
	entries := make([]*catalogEntry, 0, len(r.catalog))
	for _, entry := range r.catalog {
		entries = append(entries, entry)
	}
	r.mu.Unlock()
	sort.Strings(requestIDs)
	for _, requestID := range requestIDs {
		ctx, cancel := context.WithTimeout(context.Background(), defaultCleanupTimeout)
		_ = r.Cleanup(ctx, requestID)
		cancel()
	}
	r.lifetime.Lock()
	defer r.lifetime.Unlock()
	var first error
	for _, entry := range entries {
		if err := entry.root.Close(); err != nil && first == nil {
			first = err
		}
		if err := entry.directory.Close(); err != nil && first == nil {
			first = err
		}
	}
	return first
}

func (r *Runtime) withRequest(requestID string, fn func(*Request) Result) Result {
	r.lifetime.RLock()
	defer r.lifetime.RUnlock()
	req, err := r.Request(requestID)
	if err != nil {
		return failureFor(err)
	}
	return fn(req)
}

func (r *Runtime) allows(requestID string, operation iop.WorkspaceOperation) (*Request, Result) {
	request, err := r.Request(requestID)
	if err != nil {
		return nil, failureFor(err)
	}
	if _, ok := request.operations[operation]; !ok {
		return nil, Result{Status: iop.WorkspaceStatus_WORKSPACE_STATUS_UNSUPPORTED, Code: iop.WorkspaceErrorCode_WORKSPACE_ERROR_CODE_UNSUPPORTED}
	}
	return request, Result{Status: iop.WorkspaceStatus_WORKSPACE_STATUS_SUCCESS}
}

func validRequestID(value string) bool {
	if len(value) == 0 || len(value) > 128 {
		return false
	}
	for i, c := range value {
		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '-' || c == '_' {
			if i == 0 && (c == '-' || c == '_') {
				return false
			}
			continue
		}
		return false
	}
	return true
}

func fileIdentity(info fs.FileInfo) (uint64, uint64, bool) {
	return platformFileIdentity(info)
}