java/org/apache/catalina/valves/PersistentValve.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.catalina.valves;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;

import jakarta.servlet.AsyncContext;
import jakarta.servlet.AsyncEvent;
import jakarta.servlet.AsyncListener;
import jakarta.servlet.ServletException;
import jakarta.servlet.http.HttpServletResponse;

import org.apache.catalina.Container;
import org.apache.catalina.Context;
import org.apache.catalina.Engine;
import org.apache.catalina.Host;
import org.apache.catalina.Manager;
import org.apache.catalina.Session;
import org.apache.catalina.Store;
import org.apache.catalina.StoreManager;
import org.apache.catalina.connector.Request;
import org.apache.catalina.connector.Response;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;

/**
 * Valve that implements per-request session persistence. It is intended to be used with non-sticky load-balancers and a
 * PersistentManager. The Valve works by loading the session from the Store at the start of the request, the request
 * then updates the session as required and the Valve saves the session to the Store at the end of the request.
 * <p>
 * To avoid conflicts and/or errors when updating the session store, each session must only be accessed by no more than
 * one concurrent request. The {@code filter} field can be used to define requests (e.g. those for static resources)
 * that do not need access to the session and can bypass the session load/save functionality provided by this Valve.
 * <p>
 * The Valve uses a per session {@code Semaphore} to ensure that each session is accessed by no more than one request at
 * a time within a single Tomcat instance. The behaviour if multiple requests try to access the session concurrently can
 * be controlled by the {@code semaphoreFairness}, {@code semaphoreBlockOnAcquire} and {@code
 * semaphoreAcquireUninterruptibly} fields. If a request fails to obtain the Semaphore, the response is generated by the
 * {@link #onSemaphoreNotAcquired(Request, Response)} method which, by default, returns a {@code 429} status code.
 * <p>
 * The per session Semaphores only provide limited protection against concurrent requests within a single Tomcat
 * instance. If multiple requests access the same session concurrently across different Tomcat instances, update
 * conflicts and/or session data loss and/or errors are very likely.
 * <p>
 * <b>USAGE CONSTRAINTS</b>:
 * <ul>
 * <li>This Valve must only be used with a PersistentManager</li>
 * <li>The client must ensure that no more than one concurrent request accesses a session at any time across all Tomcat
 * instances</li>
 * </ul>
 */
public class PersistentValve extends ValveBase {

    private static final Log log = LogFactory.getLog(PersistentValve.class);

    // Saves a couple of calls to getClassLoader() on every request. Under high
    // load these calls took just long enough to appear as a hot spot (although
    // a very minor one) in a profiler.
    private static final ClassLoader MY_CLASSLOADER = PersistentValve.class.getClassLoader();

    private volatile boolean clBindRequired;

    /**
     * Regular expression pattern used to filter requests that should bypass session persistence.
     */
    protected Pattern filter = null;

    private final ConcurrentMap<String,UsageCountingSemaphore> sessionToSemaphoreMap = new ConcurrentHashMap<>();

    private boolean semaphoreFairness = true;

    private boolean semaphoreBlockOnAcquire = true;

    private boolean semaphoreAcquireUninterruptibly = true;


    /**
     * Constructs a new PersistentValve instance.
     */
    public PersistentValve() {
        super(true);
    }


    @Override
    public void setContainer(Container container) {
        super.setContainer(container);
        clBindRequired = container instanceof Engine || container instanceof Host;
    }


    @Override
    public void invoke(Request request, Response response) throws IOException, ServletException {

        // request without session
        if (isRequestWithoutSession(request.getDecodedRequestURI())) {
            if (containerLog.isTraceEnabled()) {
                containerLog.trace(sm.getString("persistentValve.requestIgnore", request.getDecodedRequestURI()));
            }
            getNext().invoke(request, response);
            return;
        } else if (containerLog.isTraceEnabled()) {
            containerLog.trace(sm.getString("persistentValve.requestProcess", request.getDecodedRequestURI()));
        }

        // Select the Context to be used for this Request
        Context context = request.getContext();
        if (context == null) {
            response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR, sm.getString("standardHost.noContext"));
            return;
        }

        boolean asyncOnEntry = request.isAsync();

        String sessionId = request.getRequestedSessionId();
        UsageCountingSemaphore semaphore = null;
        boolean mustReleaseSemaphore = true;

        try {
            /*
             * If the request was in asynchronous mode when it entered the Valve, the semaphore was acquired during the
             * original request where asynchronous processing started and does not need to be acquired again.
             */
            if (!asyncOnEntry) {
                /*
                 * Acquire the per session semaphore.
                 */
                if (sessionId != null) {
                    semaphore = sessionToSemaphoreMap.compute(sessionId, (k,
                            v) -> v == null ? new UsageCountingSemaphore(semaphoreFairness) : v.incrementUsageCount());
                    if (semaphoreBlockOnAcquire) {
                        if (semaphoreAcquireUninterruptibly) {
                            semaphore.acquireUninterruptibly();
                        } else {
                            try {
                                semaphore.acquire();
                            } catch (InterruptedException e) {
                                mustReleaseSemaphore = false;
                                onSemaphoreNotAcquired(request, response);
                                if (containerLog.isDebugEnabled()) {
                                    containerLog.debug(sm.getString("persistentValve.acquireInterrupted",
                                            request.getDecodedRequestURI()));
                                }
                                return;
                            }
                        }
                    } else {
                        if (!semaphore.tryAcquire()) {
                            mustReleaseSemaphore = false;
                            onSemaphoreNotAcquired(request, response);
                            if (containerLog.isDebugEnabled()) {
                                containerLog.debug(
                                        sm.getString("persistentValve.acquireFailed", request.getDecodedRequestURI()));
                            }
                            return;
                        }
                    }
                }

                // Update the session last access time for our session (if any)
                Manager manager = context.getManager();
                if (sessionId != null && manager instanceof StoreManager) {
                    Store store = ((StoreManager) manager).getStore();
                    if (store != null) {
                        Session session = null;
                        try {
                            session = store.load(sessionId);
                        } catch (Exception e) {
                            containerLog.error(sm.getString("persistentValve.sessionLoadFail", sessionId));
                        }
                        if (session != null) {
                            if (!session.isValid() || isSessionStale(session, System.currentTimeMillis())) {
                                if (containerLog.isTraceEnabled()) {
                                    containerLog.trace("session swapped in is invalid or expired");
                                }
                                session.expire();
                                store.remove(sessionId);
                            } else if (!session.getIdInternal().equals(sessionId)) {
                                if (containerLog.isTraceEnabled()) {
                                    containerLog.trace("session swapped in has wrong session ID");
                                }
                                store.remove(sessionId);
                            } else {
                                session.setManager(manager);
                                // session.setId(sessionId); Only if new ???
                                manager.add(session);
                                // ((StandardSession)session).activate();
                                session.access();
                                session.endAccess();
                            }
                        }
                    }
                }
            }
            if (containerLog.isTraceEnabled()) {
                containerLog.trace("sessionId: " + sessionId);
            }

            // Ask the next valve to process the request.
            getNext().invoke(request, response);
        } finally {
            if (request.isAsync()) {
                /*
                 * Need to continue to hold the semaphore until asynchronous processing is complete. Register a listener
                 * that will release the Semaphore once asynchronous processing is complete. Also need to delay session
                 * persistence until completion of the asynchronous processing.
                 *
                 * The listener must only be added once so it is only added when the request was not in asynchronous
                 * mode on entry.
                 */
                if (!asyncOnEntry) {
                    AsyncContext asyncContext = request.getAsyncContext();
                    asyncContext.addListener(new StoreSessionAsyncListener(request, context, sessionId, semaphore,
                            mustReleaseSemaphore));
                }
            } else {
                storeSession(request, context, sessionId, semaphore, mustReleaseSemaphore);
            }
        }
    }


    private void storeSession(Request request, Context context, String originalSessionId,
            UsageCountingSemaphore semaphore, boolean mustReleaseSemaphore) {
        try {
            Manager manager = context.getManager();
            Session hsess;
            try {
                hsess = request.getSessionInternal(false);
            } catch (Exception e) {
                hsess = null;
            }
            String newsessionId = null;
            if (hsess != null) {
                newsessionId = hsess.getIdInternal();
            }

            if (containerLog.isTraceEnabled()) {
                containerLog.trace("newsessionId: " + newsessionId);
            }
            if (newsessionId != null) {
                try {
                    bind(context);

                    /* store the session and remove it from the manager */
                    if (manager instanceof StoreManager) {
                        try {
                            Session session = manager.findSession(newsessionId);
                            Store store = ((StoreManager) manager).getStore();
                            boolean stored = false;
                            if (session != null) {
                                if (store != null && session.isValid() &&
                                        !isSessionStale(session, System.currentTimeMillis())) {
                                    store.save(session);
                                    ((StoreManager) manager).removeSuper(session);
                                    session.recycle();
                                    stored = true;
                                }
                            }
                            if (!stored) {
                                if (containerLog.isTraceEnabled()) {
                                    containerLog.trace(
                                            "newsessionId store: " + store + " session: " + session + " valid: " +
                                                    (session == null ? "N/A" : Boolean.toString(session.isValid())) +
                                                    " stale: " + isSessionStale(session, System.currentTimeMillis()));
                                }
                            }
                        } catch (IOException ioe) {
                            containerLog.warn(sm.getString("persistentValve.sessionSaveFail", newsessionId));
                        }
                    } else {
                        if (containerLog.isTraceEnabled()) {
                            containerLog.trace("newsessionId Manager: " + manager);
                        }
                    }
                } finally {
                    unbind(context);
                }
            }
        } finally {
            if (semaphore != null) {
                if (mustReleaseSemaphore) {
                    semaphore.release();
                }
                sessionToSemaphoreMap.computeIfPresent(originalSessionId,
                        (k, v) -> v.decrementAndGetUsageCount() == 0 ? null : v);
            }
        }
    }


    /**
     * Handle the case where a semaphore cannot be obtained. The default behaviour is to return a 429 (too many
     * requests) status code.
     *
     * @param request  The request that will not be processed
     * @param response The response that will be used for this request
     *
     * @throws IOException If an I/O error occurs while working with the request or response
     */
    protected void onSemaphoreNotAcquired(Request request, Response response) throws IOException {
        response.sendError(HttpServletResponse.SC_TOO_MANY_REQUESTS);
    }


    /**
     * Indicate whether the session has been idle for longer than its expiration date as of the supplied time.
     *
     * @param session The session to check
     * @param timeNow The current time to check for
     *
     * @return <code>true</code> if the session is past its expiration
     */
    protected boolean isSessionStale(Session session, long timeNow) {

        if (session != null) {
            int maxInactiveInterval = session.getMaxInactiveInterval();
            if (maxInactiveInterval > 0) {
                long timeIdle = session.getIdleTimeInternal() / 1000L;
                return timeIdle >= maxInactiveInterval;
            }
        }

        return false;
    }


    private void bind(Context context) {
        if (clBindRequired) {
            context.bind(MY_CLASSLOADER);
        }
    }


    private void unbind(Context context) {
        if (clBindRequired) {
            context.unbind(MY_CLASSLOADER);
        }
    }

    /**
     * Determines whether the given URI should bypass session persistence based on the configured filter.
     *
     * @param uri the request URI to check
     *
     * @return {@code true} if the request should bypass session persistence, otherwise {@code false}
     */
    protected boolean isRequestWithoutSession(String uri) {
        Pattern f = filter;
        return f != null && f.matcher(uri).matches();
    }

    /**
     * Returns the filter pattern used to identify requests that should bypass session persistence.
     *
     * @return the filter pattern, or {@code null} if no filter is configured
     */
    public String getFilter() {
        if (filter == null) {
            return null;
        }
        return filter.toString();
    }

    /**
     * Sets the filter pattern used to identify requests that should bypass session persistence.
     *
     * @param filter the regular expression pattern, or {@code null} to disable filtering
     */
    public void setFilter(String filter) {
        if (filter == null || filter.isEmpty()) {
            this.filter = null;
        } else {
            try {
                this.filter = Pattern.compile(filter);
            } catch (PatternSyntaxException pse) {
                log.error(sm.getString("persistentValve.filter.failure", filter), pse);
            }
        }
    }


    /**
     * If multiple threads attempt to acquire the same per session Semaphore, will permits be granted in the same order
     * they were requested?
     *
     * @return {@code true} if fairness is enabled, otherwise {@code false}
     */
    public boolean isSemaphoreFairness() {
        return semaphoreFairness;
    }


    /**
     * Configure whether the per session Semaphores will handle granting of permits in the same order they were
     * requested if multiple threads attempt to acquire the same Semaphore.
     *
     * @param semaphoreFairness {@code true} if permits should be granted in the same order they are requested,
     *                              otherwise {@code false}
     */
    public void setSemaphoreFairness(boolean semaphoreFairness) {
        this.semaphoreFairness = semaphoreFairness;
    }


    /**
     * If a thread attempts to acquire the per session Semaphore while it is being used by another request, should the
     * thread block to wait for the Semaphore or should the request be rejected?
     *
     * @return {@code true} if the thread should block, otherwise {@code false} to reject the concurrent request
     */
    public boolean isSemaphoreBlockOnAcquire() {
        return semaphoreBlockOnAcquire;
    }


    /**
     * Configure whether a thread should block and wait for the per session Semaphore or reject the request if the
     * Semaphore is being used by another request.
     *
     * @param semaphoreBlockOnAcquire {@code true} to block, otherwise {@code false}
     */
    public void setSemaphoreBlockOnAcquire(boolean semaphoreBlockOnAcquire) {
        this.semaphoreBlockOnAcquire = semaphoreBlockOnAcquire;
    }


    /**
     * If a thread is blocking to acquire a per session Semaphore, can that thread be interrupted?
     *
     * @return {@code true} if the thread can <b>not</b> be interrupted, otherwise {@code false}.
     */
    public boolean isSemaphoreAcquireUninterruptibly() {
        return semaphoreAcquireUninterruptibly;
    }


    /**
     * Configure whether a thread blocking to acquire a per session Semaphore can be interrupted.
     *
     * @param semaphoreAcquireUninterruptibly {@code true} if the thread can <b>not</b> be interrupted, otherwise
     *                                            {@code false}.
     */
    public void setSemaphoreAcquireUninterruptibly(boolean semaphoreAcquireUninterruptibly) {
        this.semaphoreAcquireUninterruptibly = semaphoreAcquireUninterruptibly;
    }


    /*
     * The PersistentValve uses a per session semaphore to ensure that only one request accesses a session at a time. To
     * limit the size of the session ID to Semaphore map, the Semaphores are created when required and destroyed (made
     * eligible for GC) as soon as they are not required. Tracking usage in a thread-safe way requires a usage counter
     * that does not block. The Semaphore's internal tracking can't be used because the only way to increment usage is
     * via the acquire methods and they block. Therefore, this class was created which uses a separate AtomicLong long
     * to track usage.
     */
    private static class UsageCountingSemaphore {
        private final AtomicLong usageCount = new AtomicLong(1);
        private final Semaphore semaphore;

        private UsageCountingSemaphore(boolean fairness) {
            semaphore = new Semaphore(1, fairness);
        }

        private UsageCountingSemaphore incrementUsageCount() {
            usageCount.incrementAndGet();
            return this;
        }

        private long decrementAndGetUsageCount() {
            return usageCount.decrementAndGet();
        }

        private void acquire() throws InterruptedException {
            semaphore.acquire();
        }

        private void acquireUninterruptibly() {
            semaphore.acquireUninterruptibly();
        }

        private boolean tryAcquire() {
            return semaphore.tryAcquire();
        }

        private void release() {
            semaphore.release();
        }
    }


    private class StoreSessionAsyncListener implements AsyncListener {

        private final Request request;
        private final Context context;
        private final String originalSessionId;
        private final UsageCountingSemaphore semaphore;
        private final boolean mustReleaseSemaphore;

        StoreSessionAsyncListener(Request request, Context context, String originalSessionId,
                UsageCountingSemaphore semaphore, boolean mustReleaseSemaphore) {
            this.request = request;
            this.context = context;
            this.originalSessionId = originalSessionId;
            this.semaphore = semaphore;
            this.mustReleaseSemaphore = mustReleaseSemaphore;
        }


        @Override
        public void onComplete(AsyncEvent event) throws IOException {
            storeSession(request, context, originalSessionId, semaphore, mustReleaseSemaphore);
        }

        @Override
        public void onTimeout(AsyncEvent event) throws IOException {
            // NO-OP.
        }

        @Override
        public void onError(AsyncEvent event) throws IOException {
            // NO-OP.
        }

        @Override
        public void onStartAsync(AsyncEvent event) throws IOException {
            event.getAsyncContext().addListener(this);
        }
    }
}