001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.juneau.commons.concurrent; 018 019import static org.apache.juneau.commons.utils.Shorts.*; 020 021import java.time.*; 022import java.util.concurrent.*; 023import java.util.concurrent.atomic.*; 024 025/** 026 * Built-in default {@link ReplayCache}: a per-process, {@link ConcurrentHashMap}-backed seen-identifier set with 027 * self-eviction bounded by each record's own {@code expiresAtMs}. 028 * 029 * <p> 030 * <b>Not shareable across process instances by design.</b> Two separate JVMs (or two instances behind a load 031 * balancer) each holding their own {@link InMemoryReplayCache} will not see each other's recorded identifiers, so 032 * an identifier consumed against one instance and then submitted again against the other is <b>not</b> caught. 033 * Consumers who need cross-node single-use enforcement must supply a {@link ReplayCache} backed by a store shared 034 * across every node instead of relying on this default. 035 * 036 * <p> 037 * <b>Memory bound (size honestly).</b> An identifier is retained only until its own {@code expiresAtMs}, so the 038 * map's steady-state size is bounded by the product of the <i>distinct-identifier submission rate</i> and the 039 * identifier lifetime — i.e. roughly the number of distinct, still-unexpired identifiers outstanding at 040 * once, plus at most one lifetime's worth of already-expired records awaiting the next sweep. Under a high rate 041 * with a long lifetime this is not inherently small; consumers should size accordingly, or supply a store-backed 042 * {@link ReplayCache} with server-side expiry. There is deliberately <b>no</b> hard cap that drops 043 * still-unexpired identifiers: doing so would silently weaken replay detection by letting an evicted-but-valid 044 * identifier be submitted again. 045 * 046 * <p> 047 * <b>Eviction is throttled.</b> Sweeping out already-expired records is an O(n) scan, so it runs at most once per 048 * {@link #DEFAULT_SWEEP_INTERVAL_MS} rather than on every call (which would make {@link #checkAndRecord} 049 * effectively O(n) per invocation under load). The sweep exists purely to bound memory. 050 * 051 * <p> 052 * <b>Thread-safety.</b> {@link #checkAndRecord(String, long)} is safe for concurrent invocation: the 053 * check-and-record step is a single {@link ConcurrentHashMap#putIfAbsent} call, so of any two concurrent calls 054 * with the same identifier, exactly one observes a first-seen (<jk>true</jk>) outcome. The throttled sweep is 055 * guarded by an atomic compare-and-set so at most one thread sweeps per interval. 056 * 057 * @since 10.0.0 058 */ 059public class InMemoryReplayCache implements ReplayCache { 060 061 /** Default minimum interval, in milliseconds, between opportunistic expired-record sweeps. */ 062 public static final long DEFAULT_SWEEP_INTERVAL_MS = 1000L; 063 064 private final ConcurrentHashMap<String,Long> seen = new ConcurrentHashMap<>(); 065 private final Clock clock; 066 private final long sweepIntervalMs; 067 private final AtomicLong nextSweepAtMs = new AtomicLong(); 068 069 /** 070 * Constructor using the {@link #DEFAULT_SWEEP_INTERVAL_MS default} sweep interval. 071 */ 072 public InMemoryReplayCache() { 073 this(DEFAULT_SWEEP_INTERVAL_MS); 074 } 075 076 /** 077 * Constructor. 078 * 079 * @param sweepIntervalMs The minimum interval, in milliseconds, between opportunistic expired-record sweeps. 080 * {@code 0} sweeps on every call (useful in tests that assert eviction behavior). Must be {@code >= 0}. 081 */ 082 InMemoryReplayCache(long sweepIntervalMs) { 083 this(Clock.systemUTC(), sweepIntervalMs); 084 } 085 086 /** 087 * Constructor allowing the wall-clock time source to be overridden. 088 * 089 * <p> 090 * Tests that need deterministic expiry pass a {@link Clock#fixed fixed} or {@link Clock#offset offset} clock 091 * here instead of racing real time. 092 * 093 * @param clock The clock read for the current time when deciding which records have expired. Must not be <jk>null</jk>. 094 * @param sweepIntervalMs The minimum interval, in milliseconds, between opportunistic expired-record sweeps. 095 * {@code 0} sweeps on every call. Must be {@code >= 0}. 096 */ 097 InMemoryReplayCache(Clock clock, long sweepIntervalMs) { 098 if (sweepIntervalMs < 0) 099 throw new IllegalArgumentException("sweepIntervalMs " + sweepIntervalMs + " must be >= 0"); 100 this.clock = reqnn("clock", clock); 101 this.sweepIntervalMs = sweepIntervalMs; 102 } 103 104 @Override /* ReplayCache */ 105 public boolean checkAndRecord(String id, long expiresAtMs) { 106 reqnn("id", id); 107 maybeEvictExpired(); 108 return seen.putIfAbsent(id, expiresAtMs) == null; 109 } 110 111 /** 112 * Sweeps expired records at most once per {@link #sweepIntervalMs}. The atomic compare-and-set on 113 * {@code nextSweepAtMs} both throttles the sweep and ensures a single sweeper per window; a losing thread 114 * simply skips the sweep (its own {@code putIfAbsent} still runs, so no identifier is ever missed). 115 */ 116 private void maybeEvictExpired() { 117 var now = clock.millis(); 118 var next = nextSweepAtMs.get(); 119 if (now < next) 120 return; 121 if (! nextSweepAtMs.compareAndSet(next, now + sweepIntervalMs)) 122 return; 123 seen.values().removeIf(expiresAt -> expiresAt <= now); 124 } 125}