gateway fixes

beedb-gateway/pom.xml

@@ -54,22 +54,25 @@
       <groupId>org.codehaus.mojo</groupId>
       <artifactId>exec-maven-plugin</artifactId>
       <version>3.0.0</version>
+      <configuration>
+        <mainClass>com.memcache.gateway.GatewayApplication</mainClass>
+      </configuration>
+    </plugin>
+    <plugin>
+      <groupId>org.springframework.boot</groupId>
+      <artifactId>spring-boot-maven-plugin</artifactId>
+      <version>${spring.boot.version}</version>
       <executions>
         <execution>
           <goals>
-            <goal>java</goal>
+            <goal>repackage</goal>
           </goals>
         </execution>
       </executions>
       <configuration>
-        <mainClass>com.memcache.gateway.client.BeedbClient</mainClass>
+        <mainClass>com.memcache.gateway.GatewayApplication</mainClass>
       </configuration>
     </plugin>
-    <plugin>
-      <groupId>org.springframework.boot</groupId>
-      <artifactId>spring-boot-maven-plugin</artifactId>
-      <version>${spring.boot.version}</version>
-    </plugin>
   </plugins>
 </build>
 

beedb-gateway/src/main/java/com/memcache/gateway/api/ApiExceptionHandler.java

@@ -6,7 +6,7 @@
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.RestControllerAdvice;
-
+import java.io.IOException;
 import com.memcache.gateway.client.beedbexception.BeedbException;
 import com.memcache.gateway.client.beedbexception.BeedbTimeoutException;
 import com.memcache.gateway.client.beedbexception.NoLeaderException;
@@ -39,4 +39,9 @@ else if(ex instanceof BeedbTimeoutException){
         LOGGER.error("Exception occurred in the controller", ex);
         return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body(ex.getMessage());
     }
+
+    @ExceptionHandler (IOException.class)
+    public void handleClientDisconnect(IOException ex){
+        LOGGER.debug("client disconnected: {}", ex.getMessage());
+    }
 }

beedb-gateway/src/main/java/com/memcache/gateway/api/KvController.java

@@ -4,6 +4,7 @@
 
 import java.nio.charset.StandardCharsets;
 
+import java.util.ArrayList;
 import java.util.List;
 
 import org.slf4j.Logger;
@@ -19,6 +20,7 @@
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestHeader;
 import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
 import com.memcache.gateway.client.BeedbClient;
@@ -94,8 +96,18 @@ private ResponseEntity<String> reservedKey(String key) {
      * keys the current browser happened to write itself.
      */
     @GetMapping
-    public ResponseEntity<List<KeyRegistry.Entry>> keys(){
-        return ResponseEntity.ok(keyRegistry.list());
+    public ResponseEntity<List<KeyRegistry.Entry>> keys(
+            @RequestParam(value = "limit", defaultValue = "100") int limit,
+            @RequestHeader(value = SESSION_HEADER, required = false) String sessionId){
+        List<KeyRegistry.Entry> all = keyRegistry.list();            // newest first
+        int n = Math.max(0, Math.min(limit, 200));
+        List<KeyRegistry.Entry> out = new ArrayList<>(all.subList(0, Math.min(n, all.size())));
+        if(sessionId != null){
+            for(KeyRegistry.Entry e : all){
+                if(sessionId.equals(e.sessionId()) && !out.contains(e)) out.add(e);
+            }
+        }
+        return ResponseEntity.ok(out);
     }
 
     @GetMapping ("/{key}")

beedb-gateway/src/main/java/com/memcache/gateway/api/QuotaController.java

@@ -0,0 +1,33 @@
+package com.memcache.gateway.api;
+
+import java.util.Map;
+
+import jakarta.servlet.http.HttpServletRequest;
+
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import com.memcache.gateway.kv.WriteRateLimiter;
+
+/**
+ * How many writes this client has left this hour, so the page shows the server's real count
+ * instead of guessing
+ */
+@RestController
+@RequestMapping("/api")
+public class QuotaController {
+
+    private final WriteRateLimiter rateLimiter;
+
+    public QuotaController(WriteRateLimiter rateLimiter) {
+        this.rateLimiter = rateLimiter;
+    }
+
+    @GetMapping("/quota")
+    public Map<String, Integer> quota(HttpServletRequest request) {
+        String client = request.getRemoteAddr();
+        return Map.of("remaining", rateLimiter.remaining(client),
+                      "limit", rateLimiter.maxWritesPerHour());
+    }
+}

beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbClient.java

@@ -68,6 +68,27 @@ public String delete(String key){
     }
 
 
+    /*
+     * Nodes whose last stats request failed. The status poll asks every node once a second,
+     * so a node that is down for a 15-second chaos round used to log the same WARN 15 times.
+     * Logging only the two transitions -- gone, then back -- says the same thing.
+     */
+    private final java.util.Set<String> unreachableNodes = java.util.concurrent.ConcurrentHashMap.newKeySet();
+
+    private void statsFailed(String operation, String address, Exception e){
+        if(unreachableNodes.add(address)){
+            logFailure(operation, address, e);
+        }else{
+            LOGGER.debug("{} still failing on node {} ({})", operation, address, e.toString());
+        }
+    }
+
+    private void statsAnswered(String address){
+        if(unreachableNodes.remove(address)){
+            LOGGER.info("node {} is answering again", address);
+        }
+    }
+
     public NodeStatus getStats(String nodeId, String address){
         String response = null;
         BeedbConnection connection = null;
@@ -78,9 +99,10 @@ public NodeStatus getStats(String nodeId, String address){
             if(response == null) {
                 return NodeStatus.unreachable(nodeId, address);
             }
-            else return createNodeStatus(response , nodeId , address);
+            statsAnswered(address);
+            return createNodeStatus(response , nodeId , address);
         }catch(Exception e){
-            logFailure("stats", address, e);
+            statsFailed("stats", address, e);
             pool.closeConnection(connection);
             return NodeStatus.unreachable(nodeId, address);
         }
@@ -135,6 +157,13 @@ private void logFailure(String operation, String address, Exception e){
             LOGGER.warn("{} failed: node {} is unreachable ({})", operation, address, e.getMessage());
         }else if(e instanceof java.net.SocketTimeoutException){
             LOGGER.warn("{} failed: node {} timed out ({})", operation, address, e.getMessage());
+        }else if(e instanceof NoLeaderException){
+            // Expected for the second or two an election takes -- every chaos round causes one.
+            LOGGER.warn("{} failed: no leader right now ({})", operation, e.getMessage());
+        }else if(e instanceof java.util.concurrent.TimeoutException){
+            // The status poll gives each node a fixed wait and records a slow one as unreachable
+            // for that tick 
+            LOGGER.warn("{}: node {} did not answer in time", operation, address);
         }else{
             LOGGER.error("{} failed unexpectedly against node {}", operation, address, e);
         }
@@ -198,7 +227,7 @@ public List<NodeStatus> getClusterStatus(){
                 NodeStatus result = future.get(1000, TimeUnit.MILLISECONDS);
                 nodeStatuses.add(result);
             }catch(Exception e){
-                logFailure("exception in getclusterstatus ", entries.get(i).getValue(), e);
+                statsFailed("stats poll", entries.get(i).getValue(), e);
                 nodeStatuses.add(NodeStatus.unreachable(entries.get(i).getKey() , entries.get(i).getValue()));
             }
         }

beedb-gateway/src/main/java/com/memcache/gateway/client/BeedbConnection.java

@@ -3,6 +3,7 @@
 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.IOException;
+import java.net.SocketException;
 import java.net.Socket;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -113,18 +114,18 @@ public void close(){
             LOGGER.error("something is wrong ",e);
         }
     }
-
-    public String readLine(BufferedInputStream bufferedInputStream) {
+    
+    public String readLine(BufferedInputStream bufferedInputStream) throws IOException {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         while(true){
-            try{
-                int c = bufferedInputStream.read();
-                if(c == -1 || c == '\n')break;
-                baos.write(c);
-            }catch(Exception e){
-                LOGGER.error("readLine threw exception ", e);
-                break;
+            int c = bufferedInputStream.read();
+            if(c == -1){
+                // the node closed the connection mid-reply: it died or restarted. Same outcome as
+                // a reset ,  this socket is finished and must not go back into the pool.
+                throw new SocketException("connection closed by node before the reply ended");
             }
+            if(c == '\n')break;
+            baos.write(c);
         }
         byte[] buffer = baos.toByteArray();
         if(buffer.length > 0 && buffer[buffer.length-1] == '\r')return new String(buffer, 0, buffer.length-1, StandardCharsets.UTF_8);

beedb-gateway/src/main/java/com/memcache/gateway/cluster/ClusterStatusService.java

@@ -54,6 +54,8 @@ public SseEmitter registerEmitter(){
         emitters.add(emitter);  
         try {
             emitter.send(clusterView);
+        } catch (IOException e) {
+            clientGone(emitter, e);
         } catch (Exception e) {
             logFailure(e, "failed to send cluster status");
         }
@@ -66,12 +68,19 @@ public void sendClusterStatus(){
             try {
                 emitter.send(clusterView);
                 LOGGER.debug("sent cluster status");
+            } catch (IOException e) {
+                clientGone(emitter, e);
             } catch (Exception e) {
                 emitters.remove(emitter);
                 logFailure(e , "failed to send cluster status");
             }
         });
     }
+
+    private void clientGone(SseEmitter emitter, IOException e){
+        emitters.remove(emitter);
+        LOGGER.debug("stream client disconnected ({})", e.getMessage());
+    }
 
     public void updateClusterStatus(){
         try{

beedb-gateway/src/main/java/com/memcache/gateway/demo/DemoWriter.java

@@ -42,7 +42,7 @@ public DemoWriter(BeedbClient beedbClient, KeyRegistry keyRegistry){
         this.keyRegistry = keyRegistry;
     }
 
-    @Scheduled (fixedDelay = 1000)
+    @Scheduled (fixedDelay = 2000)
     public void write(){
         long n = attempts++;
         String key = KEY_PREFIX + (n % RING_SIZE);

beedb-server/src/main/java/com/memcache/Server.java

@@ -6,7 +6,6 @@
  */
 
 package com.memcache;
-import java.io.ByteArrayOutputStream;
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -55,6 +54,8 @@ public class Server implements AutoCloseable{
     private RaftTransport raftTransport;
     private static final int INITIAL_BUFFER_SIZE = 4 * 1024;
     private static final int MAX_BUFFER_SIZE = 8 * 1024;
+    // memcached: an expiry above 30 days is an absolute Unix time, not seconds-from-now.
+    private static final long MAX_RELATIVE_EXPIRY_SECONDS = 30L * 24 * 60 * 60;
     public static void main(String[] args) throws IOException{
         // read file name
         String fileName = args[0];
@@ -114,6 +115,19 @@ private static int intProperty(Properties properties, String key, int defaultVal
         return Integer.parseInt(raw.trim());
     }
 
+    /**
+     * the client's memcached expiry, as an absolute deadline in epoch milliseconds.
+     *   0              never expires              -> 0
+     *   1 .. 30 days   seconds from now           -> now + seconds
+     *   above 30 days  an absolute Unix time      -> that time, in milliseconds
+     *   negative       already expired            -> a moment in the past
+     */
+    static long absoluteExpiryMillis(long expiry, long nowMillis){
+        if(expiry == 0) return 0;
+        if(expiry > MAX_RELATIVE_EXPIRY_SECONDS) return expiry * 1000L;
+        return nowMillis + expiry * 1000L;
+    }
+
     public void start(ArrayList<String> peers, String nodeId, int clientPort,int raftPort, String stateDir, String snapshotDir, 
         String tmpDir, String walDir, int snapShotLimit, int snapShotThreshold, int minElectionTimeout, int maxElectionTimeout, int heartbeatInterval , int peerRetryBackoffInitialMs, int peerRetryBackoffMaxMs) throws IOException, InterruptedException{
         // selector to notify about new connections
@@ -380,6 +394,7 @@ else if(command.getType() == CommandType.GET){
                 return response.toProtocolString().getBytes();
             }
             else{
+                command.setExpiry(absoluteExpiryMillis(command.getExpiry(), System.currentTimeMillis())); 
                 LOGGER.info("command type is {}", command.getType());
                 LOGGER.info("command is {}", command);
                 // we have to propose this command to raft node

beedb-server/src/main/java/com/memcache/cache/Cache.java

@@ -23,7 +23,7 @@ public CacheItem get(String key) {
         return null;
     }
 
-    public void put(String key, byte[] value, int flags, int expiry) {
+    public void put(String key, byte[] value, int flags, long expiry) {
         map.put(key, new CacheItem(key, value, flags, expiry));
     }
 

beedb-server/src/main/java/com/memcache/cache/CacheItem.java

@@ -12,11 +12,14 @@ public class CacheItem {
     private boolean     infiniteExpiry;
     private int         flags;
 
-    public CacheItem(String key, byte[] value, int flags, int expiry) {
+    public CacheItem(String key, byte[] value, int flags, long expiry) {
         this.key                = key;
         this.value              = value;
         this.flags              = flags;
-        this.expiresAt          = (expiry == 0) ? -1 : System.currentTimeMillis() + (expiry * 1000L);
+        // expiry is already an absolute epoch-millisecond deadline, set once by the leader.
+        // Never read the clock here: applying an entry must give the same result on every node
+        // and on every replay.
+        this.expiresAt          = (expiry == 0) ? -1 : expiry;
         this.infiniteExpiry     = (expiry == 0);
     }
 

beedb-server/src/main/java/com/memcache/command/Command.java

@@ -7,11 +7,14 @@ public class Command {
     private CommandType type;
     private String      key;
     private int         flags;
-    private int         expiry;
+    // From the client: memcached seconds (from now, or an absolute Unix time above 30 days).
+    // After the leader converts it (Server.absoluteExpiryMillis), and in every log entry: an
+    // absolute deadline in epoch milliseconds, or 0 for "never".
+    private long        expiry;
     private int         byteLength;
     private byte[]      value;
 
-    public Command(CommandType type, String key, int flags, int expiry, int byteLength) {
+    public Command(CommandType type, String key, int flags, long expiry, int byteLength) {
         this.type      = type;
         this.key       = key;
         this.flags     = flags;
@@ -31,7 +34,7 @@ public int getFlags() {
         return flags;
     }
 
-    public int getExpiry() {
+    public long getExpiry() {
         return expiry;
     }
 
@@ -55,7 +58,7 @@ public void setFlags(int flags) {
         this.flags = flags;
     }
 
-    public void setExpiry(int expiry) {
+    public void setExpiry(long expiry) {
         this.expiry = expiry;
     }
 
@@ -69,7 +72,7 @@ public static Command deserialize(String s) {
         CommandType type = CommandType.valueOf(parts[0]);
         String key = parts[1];
         int flags = Integer.parseInt(parts[2]);
-        int expiry = Integer.parseInt(parts[3]);
+        long expiry = Long.parseLong(parts[3]);
         int byteLength = Integer.parseInt(parts[4]);
         Command cmd = new Command(type, key, flags, expiry, byteLength);
         if (parts.length == 6 && !parts[5].isEmpty()) {