Session Queuing

/**
 * Abstract class for queuing new session requests when nodes are busy
 */
abstract class NewSessionQueue implements HasReadyState, Routable {
    /** Protected constructor with tracer and registration secret */
    protected NewSessionQueue(Tracer tracer, Secret registrationSecret);
    
    /** Fast-path to detect if the queue is empty */
    abstract boolean peekEmpty();
    
    /** Add a session request to the queue */
    abstract HttpResponse addToQueue(SessionRequest request);
    
    /** Retry adding a request to the queue */
    abstract boolean retryAddToQueue(SessionRequest request);
    
    /** Remove a specific request from the queue */
    abstract Optional<SessionRequest> remove(RequestId reqId);
    
    /** Get requests that match available node stereotypes */
    abstract List<SessionRequest> getNextAvailable(Map<Capabilities, Long> stereotypes);
    
    /** Complete a request with success or failure result */
    abstract boolean complete(RequestId reqId, Either<SessionNotCreatedException, CreateSessionResponse> result);
    
    /** Clear all pending requests from the queue */
    abstract int clearQueue();
    
    /** Get current queue contents */
    abstract List<SessionRequestCapability> getQueueContents();
}

/**
 * In-memory new session queue implementation
 */
class LocalNewSessionQueue extends NewSessionQueue {
    /** Create a local session queue with event bus integration */
    LocalNewSessionQueue(Tracer tracer, Duration requestTimeout, Duration retryPeriod);
    
    /** Factory method to create from configuration */
    static NewSessionQueue create(Config config);
    
    boolean isReady(); 
    boolean offerLast(SessionRequest request, RequestId requestId);
    Optional<SessionRequest> poll(Duration timeout);
    int clear();
    List<SessionRequest> getNextAvailable(Map<Capabilities, Long> stereotypes);
    
    /** Get current queue size */
    int getQueueSize();
    
    /** Get queue statistics */
    QueueStatistics getStatistics();
}

// Create local session queue
NewSessionQueue sessionQueue = new LocalNewSessionQueue(
    tracer,
    Duration.ofMinutes(5), // request timeout
    Duration.ofSeconds(5)  // retry period
);

// Add session request to queue
SessionRequest request = new SessionRequest(
    new RequestId(UUID.randomUUID()),
    Instant.now().plus(Duration.ofMinutes(5)), // enqueued time + timeout
    Set.of(W3C), // WebDriver dialects
    new ImmutableCapabilities("browserName", "chrome")
);

boolean queued = sessionQueue.offerLast(request, request.getRequestId());
if (queued) {
    System.out.println("Request queued: " + request.getRequestId());
}

// Poll for next request (used by distributor)
Optional<SessionRequest> next = sessionQueue.poll(Duration.ofSeconds(10));
if (next.isPresent()) {
    System.out.println("Processing request: " + next.get().getRequestId());
}

/**
 * Remote session queue client for distributed deployments
 */
class RemoteNewSessionQueue extends NewSessionQueue {
    RemoteNewSessionQueue(HttpClient.Factory httpClientFactory, URI queueUri);
    
    // All operations implemented via HTTP calls
    boolean isReady();
    boolean offerLast(SessionRequest request, RequestId requestId);
    Optional<SessionRequest> poll(Duration timeout);
    int clear();
    List<SessionRequest> getNextAvailable(Map<Capabilities, Long> stereotypes);
}

/**
 * Configuration options for session queue behavior
 */
class NewSessionQueueOptions {
    static final String SESSION_QUEUE_SECTION = "sessionqueue";
    
    /** Get session queue implementation class */
    String getSessionQueueImplementation(Config config);
    
    /** Get request timeout duration */
    Duration getRequestTimeout(Config config);
    
    /** Get retry period for polling requests */
    Duration getRetryPeriod(Config config);
    
    /** Get maximum queue size */
    int getMaxQueueSize(Config config);
    
    /** Get queue cleanup interval */
    Duration getCleanupInterval(Config config);
}

// 1. Router receives new session request
@POST
@Path("/session")
public Response createSession(NewSessionPayload payload) {
    SessionRequest sessionRequest = new SessionRequest(
        new RequestId(UUID.randomUUID()),
        Instant.now().plus(requestTimeout),
        payload.getDownstreamDialects(),
        payload.getDesiredCapabilities()
    );
    
    // 2. Try immediate session creation
    Either<SessionNotCreatedException, CreateSessionResponse> result = 
        distributor.newSession(sessionRequest);
    
    if (result.isRight()) {
        // Session created immediately
        return Response.ok(result.right()).build();
    }
    
    // 3. Queue the request if nodes are busy
    boolean queued = sessionQueue.offerLast(sessionRequest, sessionRequest.getRequestId());
    if (!queued) {
        return Response.status(503).entity("Queue full").build();
    }
    
    // 4. Wait for session creation
    return waitForSession(sessionRequest.getRequestId());
}

// Distributor processes queued requests
public class QueueProcessingDistributor extends LocalDistributor {
    @Scheduled(fixedRate = 1000) // Every second
    public void processQueue() {
        // Get available node capabilities
        Map<Capabilities, Long> availableSlots = getAvailableSlots();
        
        // Get matching requests from queue
        List<SessionRequest> availableRequests = 
            sessionQueue.getNextAvailable(availableSlots);
        
        for (SessionRequest request : availableRequests) {
            Either<SessionNotCreatedException, CreateSessionResponse> result = 
                newSession(request);
            
            if (result.isRight()) {
                // Notify waiting client
                notifySessionCreated(request.getRequestId(), result.right());
            } else {
                // Check if request has expired
                if (request.getEnqueued().isBefore(Instant.now().minus(requestTimeout))) {
                    notifySessionFailed(request.getRequestId(), "Request timeout");
                } else {
                    // Put back in queue for retry
                    sessionQueue.offerLast(request, request.getRequestId()); 
                }
            }
        }
    }
}

// Cleanup expired requests
@Scheduled(fixedRate = 30000) // Every 30 seconds
public void cleanupExpiredRequests() {
    Instant cutoff = Instant.now();
    List<SessionRequest> expiredRequests = new ArrayList<>();
    
    // Check all queued requests
    SessionRequest request;
    while ((request = sessionQueue.poll(Duration.ZERO)) != null) {
        if (request.getEnqueued().isAfter(cutoff)) {
            // Request still valid, put back in queue
            sessionQueue.offerLast(request, request.getRequestId());
        } else {
            // Request expired
            expiredRequests.add(request);
        }
    }
    
    // Notify clients of expired requests
    for (SessionRequest expired : expiredRequests) {
        notifySessionFailed(expired.getRequestId(), "Request timeout");
    }
}

// Custom queue with priority support
public class PrioritySessionQueue implements NewSessionQueue {
    private final PriorityQueue<PrioritizedRequest> queue;
    
    static class PrioritizedRequest implements Comparable<PrioritizedRequest> {
        final SessionRequest request;
        final int priority;
        final Instant enqueued;
        
        @Override
        public int compareTo(PrioritizedRequest other) {
            // Higher priority first, then FIFO for same priority
            int priorityCompare = Integer.compare(other.priority, this.priority);
            if (priorityCompare != 0) return priorityCompare;
            return this.enqueued.compareTo(other.enqueued);
        }
    }
    
    @Override
    public boolean offerLast(SessionRequest request, RequestId requestId) {
        int priority = extractPriority(request.getDesiredCapabilities());
        return queue.offer(new PrioritizedRequest(request, priority, Instant.now()));
    }
    
    private int extractPriority(Capabilities caps) {
        // Extract priority from capabilities or use default
        return (Integer) caps.getCapability("se:priority", 0);
    }
}

// Queue that considers current grid load
public class LoadAwareSessionQueue extends LocalNewSessionQueue {
    @Override
    public List<SessionRequest> getNextAvailable(Map<Capabilities, Long> stereotypes) {
        List<SessionRequest> available = super.getNextAvailable(stereotypes);
        
        // Sort by grid load - prefer requests for less loaded browser types
        return available.stream()
            .sorted((r1, r2) -> {
                String browser1 = r1.getDesiredCapabilities().getBrowserName();
                String browser2 = r2.getDesiredCapabilities().getBrowserName();
                
                long load1 = getCurrentLoad(browser1);
                long load2 = getCurrentLoad(browser2);
                
                return Long.compare(load1, load2);
            })
            .collect(Collectors.toList());
    }
    
    private long getCurrentLoad(String browserName) {
        // Calculate current load for browser type
        return distributor.getStatus().getNodes().stream()
            .mapToLong(node -> node.getSlots().stream()
                .filter(slot -> slot.getStereotype().getBrowserName().equals(browserName))
                .filter(slot -> slot.getSession() != null)
                .count())
            .sum();
    }
}

// Queue monitoring
public class QueueMetrics {
    public int getQueueSize() {
        return sessionQueue.getQueueSize();
    }
    
    public Duration getAverageWaitTime() {
        // Track wait times for completed requests
        return averageWaitTime;
    }
    
    public Map<String, Integer> getQueuedRequestsByBrowser() {
        // Get breakdown of queued requests by browser
        return queuedRequests.stream()
            .collect(Collectors.groupingBy(
                request -> request.getDesiredCapabilities().getBrowserName(),
                Collectors.summingInt(request -> 1)
            ));
    }
    
    public int getExpiredRequestCount() {
        return expiredRequestCount.get();
    }
}

// Handle queue errors
try {
    boolean queued = sessionQueue.offerLast(request, requestId);
    if (!queued) {
        // Queue full or request rejected
        return Response.status(503)
            .entity(Map.of("error", "Unable to queue request - queue may be full"))
            .build();
    }
} catch (Exception e) {
    // Queue service unavailable
    return Response.status(503)
        .entity(Map.of("error", "Queue service unavailable: " + e.getMessage()))
        .build();
}

// Handle polling errors
try {
    Optional<SessionRequest> next = sessionQueue.poll(Duration.ofSeconds(5));
    // Process request...
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    throw new RuntimeException("Queue polling interrupted", e);
} catch (Exception e) {
    // Log error and continue processing
    log.warn("Error polling session queue", e);
}