Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 8 additions & 11 deletions server/cmd/api/api/api.go
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import (
"fmt"
"log/slog"
"os"
"os/exec"
"sync"
"time"

Expand Down Expand Up @@ -89,14 +88,9 @@ type ApiService struct {
// inputMu serializes input-related operations (mouse, keyboard, screenshot)
inputMu sync.Mutex

// playwrightMu serializes Playwright code execution (only one execution at a time)
playwrightMu sync.Mutex

// playwrightDaemonStarting is an atomic flag to prevent concurrent daemon starts
playwrightDaemonStarting int32

// playwrightDaemonCmd holds the daemon process for cleanup
playwrightDaemonCmd *exec.Cmd
// playwrightExecutors runs Playwright calls, each executor (including the
// default one) in its own daemon process.
playwrightExecutors *playwrightExecutorManager

browserRepl *browserReplManager

Expand Down Expand Up @@ -179,7 +173,7 @@ func New(
_ = mon.SetTelemetry(false)
ctx, cancel := context.WithCancel(context.Background())

return &ApiService{
s := &ApiService{
recordManager: recordManager,
factory: factory,
defaultRecorderID: "default",
Expand All @@ -198,7 +192,9 @@ func New(
browserRepl: newBrowserReplManager(),
lifecycleCtx: ctx,
lifecycleCancel: cancel,
}, nil
}
s.playwrightExecutors = newPlaywrightExecutorManager(cdpPlaywrightExecutorTabs{withCDP: s.withCDPClient})
return s, nil
}

func (s *ApiService) StartRecording(ctx context.Context, req oapi.StartRecordingRequestObject) (oapi.StartRecordingResponseObject, error) {
Expand Down Expand Up @@ -472,6 +468,7 @@ func (s *ApiService) NetworkMetrics() (resets, completed uint64, up bool) {
func (s *ApiService) Shutdown(ctx context.Context) error {
s.lifecycleCancel()
replErr := s.browserRepl.Shutdown(ctx)
s.playwrightExecutors.Shutdown(ctx)

_ = s.webmcp.Close()
s.monitorMu.Lock()
Expand Down
187 changes: 56 additions & 131 deletions server/cmd/api/api/playwright.go
Original file line number Diff line number Diff line change
@@ -1,187 +1,112 @@
package api

import (
"bufio"
"context"
"encoding/json"
"errors"
"fmt"
"net"
"os"
"os/exec"
"sync/atomic"
"time"

"github.com/google/uuid"
"github.com/kernel/kernel-images/server/lib/logger"
"github.com/kernel/kernel-images/server/lib/oapi"
)

const (
playwrightDaemonSocket = "/tmp/playwright-daemon.sock"
playwrightDaemonScript = "/usr/local/lib/playwright-daemon.js"
playwrightDaemonStartup = 5 * time.Second
)

type playwrightDaemonRequest struct {
ID string `json:"id"`
Code string `json:"code"`
TimeoutMs int `json:"timeout_ms,omitempty"`
ID string `json:"id"`
Code string `json:"code"`
TimeoutMs int `json:"timeout_ms,omitempty"`
Executor string `json:"executor,omitempty"`
TargetID string `json:"target_id,omitempty"`
TabCreated bool `json:"tab_created,omitempty"`
}

type playwrightDaemonResponse struct {
ID string `json:"id"`
Success bool `json:"success"`
Result interface{} `json:"result,omitempty"`
Error string `json:"error,omitempty"`
Stack string `json:"stack,omitempty"`
ID string `json:"id"`
Success bool `json:"success"`
Result interface{} `json:"result,omitempty"`
Error string `json:"error,omitempty"`
Stack string `json:"stack,omitempty"`
TargetID string `json:"target_id,omitempty"`
TabCreated bool `json:"tab_created,omitempty"`
TimedOut bool `json:"timed_out,omitempty"`
TabMissing bool `json:"tab_missing,omitempty"`
}

func (s *ApiService) ensurePlaywrightDaemon(ctx context.Context) error {
log := logger.FromContext(ctx)

if conn, err := net.DialTimeout("unix", playwrightDaemonSocket, 100*time.Millisecond); err == nil {
conn.Close()
return nil
}

if !atomic.CompareAndSwapInt32(&s.playwrightDaemonStarting, 0, 1) {
deadline := time.Now().Add(playwrightDaemonStartup)
for time.Now().Before(deadline) {
if conn, err := net.DialTimeout("unix", playwrightDaemonSocket, 100*time.Millisecond); err == nil {
conn.Close()
return nil
}
time.Sleep(100 * time.Millisecond)
}
return fmt.Errorf("timeout waiting for daemon to start")
func (s *ApiService) ExecutePlaywrightCode(ctx context.Context, request oapi.ExecutePlaywrightCodeRequestObject) (oapi.ExecutePlaywrightCodeResponseObject, error) {
if request.Body == nil || request.Body.Code == "" {
return oapi.ExecutePlaywrightCode400JSONResponse{
BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{
Message: "code is required",
},
}, nil
}
defer atomic.StoreInt32(&s.playwrightDaemonStarting, 0)

log.Info("starting playwright daemon")

cmd := exec.Command("node", playwrightDaemonScript)
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
cmd.Env = os.Environ()
RecordTelemetryCode(ctx, request.Body.Code)

if err := cmd.Start(); err != nil {
return fmt.Errorf("failed to start playwright daemon: %w", err)
timeout := 60 * time.Second
if request.Body.TimeoutSec != nil && *request.Body.TimeoutSec > 0 {
timeout = time.Duration(*request.Body.TimeoutSec) * time.Second
}

s.playwrightDaemonCmd = cmd

deadline := time.Now().Add(playwrightDaemonStartup)
for time.Now().Before(deadline) {
if conn, err := net.DialTimeout("unix", playwrightDaemonSocket, 100*time.Millisecond); err == nil {
conn.Close()
log.Info("playwright daemon started successfully")
return nil
}
time.Sleep(100 * time.Millisecond)
name := defaultPlaywrightExecutor
if request.Body.Executor != nil {
name = *request.Body.Executor
}

cmd.Process.Kill()
return fmt.Errorf("playwright daemon failed to start within %v", playwrightDaemonStartup)
return s.executePlaywrightOnExecutor(ctx, name, request.Body.Code, timeout)
}

func (s *ApiService) executeViaUnixSocket(ctx context.Context, code string, timeout time.Duration) (*playwrightDaemonResponse, error) {
conn, err := net.DialTimeout("unix", playwrightDaemonSocket, 2*time.Second)
if err != nil {
return nil, fmt.Errorf("failed to connect to daemon: %w", err)
}
defer conn.Close()

if err := conn.SetDeadline(time.Now().Add(timeout + 5*time.Second)); err != nil {
return nil, fmt.Errorf("failed to set deadline: %w", err)
}

reqID := uuid.New().String()
req := playwrightDaemonRequest{
ID: reqID,
Code: code,
TimeoutMs: int(timeout.Milliseconds()),
}

reqBytes, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("failed to marshal request: %w", err)
}
reqBytes = append(reqBytes, '\n')

if _, err := conn.Write(reqBytes); err != nil {
return nil, fmt.Errorf("failed to send request: %w", err)
}

reader := bufio.NewReader(conn)
respLine, err := reader.ReadBytes('\n')
if err != nil {
return nil, fmt.Errorf("failed to read response: %w", err)
func playwrightResult(resp *playwrightDaemonResponse) oapi.ExecutePlaywrightResult {
result := oapi.ExecutePlaywrightResult{Success: resp.Success}
if resp.TargetID != "" {
result.Tab = &oapi.PlaywrightTab{TargetId: resp.TargetID, Created: resp.TabCreated}
}

var resp playwrightDaemonResponse
if err := json.Unmarshal(respLine, &resp); err != nil {
return nil, fmt.Errorf("failed to parse response: %w", err)
}

if resp.ID != reqID {
return nil, fmt.Errorf("response ID mismatch: expected %s, got %s", reqID, resp.ID)
if resp.Success {
result.Result = &resp.Result
} else {
result.Error = &resp.Error
result.Stderr = &resp.Stack
}

return &resp, nil
return result
}

func (s *ApiService) ExecutePlaywrightCode(ctx context.Context, request oapi.ExecutePlaywrightCodeRequestObject) (oapi.ExecutePlaywrightCodeResponseObject, error) {
s.playwrightMu.Lock()
defer s.playwrightMu.Unlock()

func (s *ApiService) executePlaywrightOnExecutor(ctx context.Context, name, code string, timeout time.Duration) (oapi.ExecutePlaywrightCodeResponseObject, error) {
log := logger.FromContext(ctx)

if request.Body == nil || request.Body.Code == "" {
if !playwrightExecutorNamePattern.MatchString(name) {
return oapi.ExecutePlaywrightCode400JSONResponse{
BadRequestErrorJSONResponse: oapi.BadRequestErrorJSONResponse{
Message: "code is required",
Message: "executor name must match " + playwrightExecutorNamePattern.String(),
},
}, nil
}

RecordTelemetryCode(ctx, request.Body.Code)

timeout := 60 * time.Second
if request.Body.TimeoutSec != nil && *request.Body.TimeoutSec > 0 {
timeout = time.Duration(*request.Body.TimeoutSec) * time.Second
resp, err := s.playwrightExecutors.Execute(ctx, name, code, timeout)
var limitErr *playwrightExecutorLimitError
if errors.As(err, &limitErr) {
return oapi.ExecutePlaywrightCode409JSONResponse{
Message: limitErr.Error(),
Executors: s.playwrightExecutorsJSON(ctx, limitErr.executors),
}, nil
}

if err := s.ensurePlaywrightDaemon(ctx); err != nil {
log.Error("failed to ensure playwright daemon", "error", err)
if errors.Is(err, errPlaywrightExecutorSetup) {
log.Error("failed to set up playwright executor", "executor", name, "error", err)
return oapi.ExecutePlaywrightCode500JSONResponse{
InternalErrorJSONResponse: oapi.InternalErrorJSONResponse{
Message: fmt.Sprintf("failed to start playwright daemon: %v", err),
Message: err.Error(),
},
}, nil
}

resp, err := s.executeViaUnixSocket(ctx, request.Body.Code, timeout)
if err != nil {
log.Error("playwright execution failed", "error", err)
log.Error("playwright executor execution failed", "executor", name, "error", err)
errorMsg := fmt.Sprintf("execution failed: %v", err)
return oapi.ExecutePlaywrightCode200JSONResponse{
Success: false,
Error: &errorMsg,
}, nil
}

if !resp.Success {
errorMsg := resp.Error
stderr := resp.Stack
return oapi.ExecutePlaywrightCode200JSONResponse{
Success: false,
Error: &errorMsg,
Stderr: &stderr,
}, nil
}

return oapi.ExecutePlaywrightCode200JSONResponse{
Success: true,
Result: &resp.Result,
}, nil
return oapi.ExecutePlaywrightCode200JSONResponse(playwrightResult(resp)), nil
}
Loading
Loading