Various improvements/fixes on long-running stability

.github/workflows/build-cov.yaml

@@ -1,6 +1,7 @@
 name: Cov-Build
 
 on:
+  workflow_dispatch:
   pull_request:
     branches:
       - 'main'

base-cluster/pom.xml

@@ -33,6 +33,10 @@
             <groupId>org.apache.bifromq</groupId>
             <artifactId>base-env-provider</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.bifromq</groupId>
+            <artifactId>base-util</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.apache.bifromq</groupId>
             <artifactId>base-hlc</artifactId>

base-cluster/src/main/java/org/apache/bifromq/basecluster/AgentHost.java

@@ -14,14 +14,31 @@
  * "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.    
+ * under the License.
  */
 
 package org.apache.bifromq.basecluster;
 
-import static org.apache.bifromq.basecluster.memberlist.CRDTUtil.AGENT_HOST_MAP_URI;
 import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.bifromq.basecluster.memberlist.CRDTUtil.AGENT_HOST_MAP_URI;
 
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.protobuf.ByteString;
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.disposables.CompositeDisposable;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.bifromq.basecluster.agent.proto.AgentEndpoint;
 import org.apache.bifromq.basecluster.fd.FailureDetector;
 import org.apache.bifromq.basecluster.fd.IFailureDetector;
@@ -43,23 +60,6 @@
 import org.apache.bifromq.basecrdt.store.ICRDTStore;
 import org.apache.bifromq.basecrdt.store.proto.CRDTStoreMessage;
 import org.apache.bifromq.baseenv.EnvProvider;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-import com.google.protobuf.ByteString;
-import io.micrometer.core.instrument.Metrics;
-import io.micrometer.core.instrument.binder.jvm.ExecutorServiceMetrics;
-import io.reactivex.rxjava3.core.Observable;
-import io.reactivex.rxjava3.core.Scheduler;
-import io.reactivex.rxjava3.disposables.CompositeDisposable;
-import io.reactivex.rxjava3.schedulers.Schedulers;
-import java.net.InetSocketAddress;
-import java.time.Duration;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.atomic.AtomicReference;
-import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 final class AgentHost implements IAgentHost {
@@ -173,6 +173,11 @@ public Observable<Map<HostEndpoint, Set<String>>> landscape() {
         return memberList.landscape();
     }
 
+    @Override
+    public Observable<Long> refuteSignal() {
+        return memberList.refuteSignal();
+    }
+
     @Override
     public void close() {
         if (state.compareAndSet(State.STARTED, State.STOPPING)) {

base-cluster/src/main/java/org/apache/bifromq/basecluster/IAgentHost.java

@@ -14,23 +14,23 @@
  * "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.    
+ * under the License.
  */
 
 package org.apache.bifromq.basecluster;
 
+import io.reactivex.rxjava3.core.Observable;
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import org.apache.bifromq.basecluster.memberlist.HostAddressResolver;
 import org.apache.bifromq.basecluster.memberlist.IHostAddressResolver;
 import org.apache.bifromq.basecluster.memberlist.agent.IAgent;
 import org.apache.bifromq.basecluster.membership.proto.HostEndpoint;
 import org.apache.bifromq.basecluster.transport.ITransport;
 import org.apache.bifromq.basecluster.transport.TCPTransport;
 import org.apache.bifromq.basecluster.transport.Transport;
-import io.reactivex.rxjava3.core.Observable;
-import java.net.InetSocketAddress;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
 
 /**
  * Agent host defines the interface for hosting agents and joining the cluster.
@@ -101,6 +101,14 @@ static IAgentHost newInstance(AgentHostOptions options) {
      */
     Observable<Map<HostEndpoint, Set<String>>> landscape();
 
+    /**
+     * Emits a signal whenever the local host actively refutes a suspicion of being dead.
+     * Each emission carries the timestamp (in millis) when the refutation occurred.
+     *
+     * @return an observable stream of refutation timestamps
+     */
+    Observable<Long> refuteSignal();
+
     /**
      * Shutdown the agent host.
      */

base-cluster/src/main/java/org/apache/bifromq/basecluster/memberlist/HostMemberList.java

@@ -14,7 +14,7 @@
  * "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.    
+ * under the License.
  */
 
 package org.apache.bifromq.basecluster.memberlist;
@@ -25,28 +25,8 @@
 import static org.apache.bifromq.basecrdt.core.api.CausalCRDTType.mvreg;
 import static org.apache.bifromq.basecrdt.store.ReplicaIdGenerator.generate;
 
-import org.apache.bifromq.basecluster.agent.proto.AgentEndpoint;
-import org.apache.bifromq.basecluster.memberlist.agent.Agent;
-import org.apache.bifromq.basecluster.memberlist.agent.AgentAddressProvider;
-import org.apache.bifromq.basecluster.memberlist.agent.AgentMessenger;
-import org.apache.bifromq.basecluster.memberlist.agent.IAgent;
-import org.apache.bifromq.basecluster.membership.proto.Doubt;
-import org.apache.bifromq.basecluster.membership.proto.Fail;
-import org.apache.bifromq.basecluster.membership.proto.HostEndpoint;
-import org.apache.bifromq.basecluster.membership.proto.HostMember;
-import org.apache.bifromq.basecluster.membership.proto.Join;
-import org.apache.bifromq.basecluster.membership.proto.Quit;
-import org.apache.bifromq.basecluster.messenger.IMessenger;
-import org.apache.bifromq.basecluster.proto.ClusterMessage;
-import org.apache.bifromq.basecrdt.core.api.IORMap;
-import org.apache.bifromq.basecrdt.core.api.MVRegOperation;
-import org.apache.bifromq.basecrdt.core.api.ORMapOperation;
-import org.apache.bifromq.basecrdt.proto.Replica;
-import org.apache.bifromq.basecrdt.store.ICRDTStore;
-import org.apache.bifromq.basehlc.HLC;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
 import com.google.protobuf.AbstractMessageLite;
 import com.google.protobuf.ByteString;
 import io.micrometer.core.instrument.Gauge;
@@ -57,6 +37,7 @@
 import io.reactivex.rxjava3.core.Scheduler;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
 import io.reactivex.rxjava3.subjects.BehaviorSubject;
+import io.reactivex.rxjava3.subjects.PublishSubject;
 import java.net.InetSocketAddress;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -69,7 +50,30 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bifromq.base.util.RendezvousHash;
+import org.apache.bifromq.basecluster.agent.proto.AgentEndpoint;
+import org.apache.bifromq.basecluster.memberlist.agent.Agent;
+import org.apache.bifromq.basecluster.memberlist.agent.AgentAddressProvider;
+import org.apache.bifromq.basecluster.memberlist.agent.AgentMessenger;
+import org.apache.bifromq.basecluster.memberlist.agent.IAgent;
+import org.apache.bifromq.basecluster.membership.proto.Doubt;
+import org.apache.bifromq.basecluster.membership.proto.Fail;
+import org.apache.bifromq.basecluster.membership.proto.HostEndpoint;
+import org.apache.bifromq.basecluster.membership.proto.HostMember;
+import org.apache.bifromq.basecluster.membership.proto.Join;
+import org.apache.bifromq.basecluster.membership.proto.Quit;
+import org.apache.bifromq.basecluster.messenger.IMessenger;
+import org.apache.bifromq.basecluster.proto.ClusterMessage;
+import org.apache.bifromq.basecrdt.core.api.IORMap;
+import org.apache.bifromq.basecrdt.core.api.MVRegOperation;
+import org.apache.bifromq.basecrdt.core.api.ORMapOperation;
+import org.apache.bifromq.basecrdt.proto.Replica;
+import org.apache.bifromq.basecrdt.store.ICRDTStore;
+import org.apache.bifromq.basehlc.HLC;
 
+/**
+ * HostMemberList implementation using CRDT for achieving a consistent view of the host members in the cluster.
+ */
 @Slf4j
 public class HostMemberList implements IHostMemberList {
     private final AtomicReference<State> state = new AtomicReference<>(State.JOINED);
@@ -79,12 +83,25 @@ public class HostMemberList implements IHostMemberList {
     private final IHostAddressResolver addressResolver;
     private final BehaviorSubject<Map<HostEndpoint, HostMember>> membershipSubject = BehaviorSubject.createDefault(
         new ConcurrentHashMap<>());
+    private final PublishSubject<Long> refuteSubject = PublishSubject.create();
     private final Map<String, Agent> agentMap = new ConcurrentHashMap<>();
     private final IORMap hostListCRDT;
     private final CompositeDisposable disposables = new CompositeDisposable();
     private final MetricManager metricManager;
     private final String[] tags;
     private volatile HostMember local;
+
+    /**
+     * Constructor of HostMemberList.
+     *
+     * @param bindAddr the address to bind the host member
+     * @param port the port to bind the host member
+     * @param messenger the messenger to use for communication
+     * @param scheduler the scheduler to use for scheduling tasks
+     * @param store the CRDT store to use for storing internal OR-Map
+     * @param addressResolver the address resolver to resolve host endpoints to addresses
+     * @param tags the tags to be used for metrics
+     */
     public HostMemberList(String bindAddr,
                           int port,
                           IMessenger messenger,
@@ -134,10 +151,13 @@ private boolean join(HostMember member) {
             if (joined) {
                 // add it into crdt
                 log.debug("Member[{}] joins the cluster: local={}", member, local);
-                Optional<HostMember> memberInCRDT = getHostMember(hostListCRDT, member.getEndpoint());
-                if (memberInCRDT.isEmpty() || memberInCRDT.get().getIncarnation() < member.getIncarnation()) {
-                    hostListCRDT.execute(ORMapOperation.update(member.getEndpoint().toByteString())
-                        .with(MVRegOperation.write(member.toByteString())));
+                if (member == local) {
+                    // only update crdt if it's local member
+                    Optional<HostMember> memberInCRDT = getHostMember(hostListCRDT, member.getEndpoint());
+                    if (memberInCRDT.isEmpty() || memberInCRDT.get().getIncarnation() < member.getIncarnation()) {
+                        hostListCRDT.execute(ORMapOperation.update(member.getEndpoint().toByteString())
+                            .with(MVRegOperation.write(member.toByteString())));
+                    }
                 }
                 // update crdt landscape
                 store.join(hostListCRDT.id(), currentMembers().keySet().stream()
@@ -148,12 +168,11 @@ private boolean join(HostMember member) {
         }
     }
 
-    private void drop(HostEndpoint memberEndpoint, int incarnation) {
+    private void drop(HostEndpoint memberEndpoint, int incarnation, boolean fromQuit) {
         synchronized (this) {
             boolean removed = removeMember(memberEndpoint, incarnation);
             Optional<HostMember> memberInCRDT = getHostMember(hostListCRDT, memberEndpoint);
-            if (memberInCRDT.isPresent()) {
-                // remove it from crdt if any
+            if (!fromQuit && memberInCRDT.isPresent() && shouldReportFailure(memberInCRDT.get().getEndpoint())) {
                 hostListCRDT.execute(ORMapOperation.remove(memberEndpoint.toByteString()).of(mvreg));
             }
             if (removed) {
@@ -165,6 +184,17 @@ private void drop(HostEndpoint memberEndpoint, int incarnation) {
         }
     }
 
+    private boolean shouldReportFailure(HostEndpoint failedMemberEndpoint) {
+        // if local member is responsible for removing the failed member from CRDT
+        RendezvousHash<HostEndpoint, HostEndpoint> hash = RendezvousHash.<HostEndpoint, HostEndpoint>builder()
+            .keyFunnel((from, into) -> into.putBytes(from.getId().asReadOnlyByteBuffer()))
+            .nodeFunnel((from, into) -> into.putBytes(from.getId().asReadOnlyByteBuffer()))
+            .nodes(currentMembers().keySet())
+            .build();
+        HostEndpoint reporter = hash.get(failedMemberEndpoint);
+        return reporter.getId().equals(local.getEndpoint().getId());
+    }
+
     @Override
     public boolean isZombie(HostEndpoint endpoint) {
         return !endpoint.getId().equals(local.getEndpoint().getId())
@@ -207,6 +237,7 @@ public CompletableFuture<Void> stop() {
                 .thenCompose(v -> store.stopHosting(hostListCRDT.id()))
                 .whenComplete((v, e) -> {
                     membershipSubject.onComplete();
+                    refuteSubject.onComplete();
                     metricManager.close();
                     state.set(State.QUITED);
                 });
@@ -226,6 +257,8 @@ private void renew(int atLeastIncarnation) {
         synchronized (this) {
             local = local.toBuilder().setIncarnation(Math.max(local.getIncarnation(), atLeastIncarnation) + 1).build();
             join(local);
+            agentMap.values().forEach(Agent::refreshRegistration);
+            refuteSubject.onNext(HLC.INST.get());
         }
     }
 
@@ -247,7 +280,6 @@ public IAgent host(String agentId) {
                     tags));
                 local = local.toBuilder()
                     .setIncarnation(local.getIncarnation() + 1)
-                    .addAgentId(agentId) // deprecate since 3.3.3
                     .putAgent(agentId, agentEndpoint.getIncarnation())
                     .build();
                 join(local);
@@ -265,8 +297,6 @@ public CompletableFuture<Void> stopHosting(String agentId) {
                 synchronized (this) {
                     local = local.toBuilder()
                         .setIncarnation(local.getIncarnation() + 1)
-                        .clearAgentId()
-                        .addAllAgentId(agentMap.keySet()) // deprecate since 3.3.3
                         .clearAgent()
                         .putAllAgent(Maps.transformValues(agentMap, a -> a.local().getIncarnation()))
                         .build();
@@ -279,7 +309,12 @@ public CompletableFuture<Void> stopHosting(String agentId) {
 
     @Override
     public Observable<Map<HostEndpoint, Set<String>>> landscape() {
-        return membershipSubject.map(m -> Maps.transformValues(m, v -> Sets.newHashSet(v.getAgentIdList())));
+        return membershipSubject.map(m -> Maps.transformValues(m, v -> v.getAgentMap().keySet()));
+    }
+
+    @Override
+    public Observable<Long> refuteSignal() {
+        return refuteSubject;
     }
 
     private Map<HostEndpoint, HostMember> currentMembers() {
@@ -327,6 +362,9 @@ private void handleMessage(ClusterMessage message) {
             case QUIT -> handleQuit(message.getQuit());
             case FAIL -> handleFail(message.getFail());
             case DOUBT -> handleDoubt(message.getDoubt());
+            default -> {
+                // never happen
+            }
         }
     }
 
@@ -363,15 +401,15 @@ private void handleFail(Fail fail) {
         } else if (isZombie(failedEndpoint)) {
             clearZombie(failedEndpoint);
         } else {
-            drop(failedEndpoint, fail.getIncarnation());
+            drop(failedEndpoint, fail.getIncarnation(), false);
         }
     }
 
     private void handleQuit(Quit quit) {
         HostEndpoint quitEndpoint = quit.getEndpoint();
         log.debug("Member[{}] quits the cluster", quitEndpoint);
         if (!quitEndpoint.equals(local.getEndpoint()) && !isZombie(quitEndpoint)) {
-            drop(quitEndpoint, quit.getIncarnation());
+            drop(quitEndpoint, quit.getIncarnation(), true);
         }
     }
 
@@ -388,7 +426,7 @@ private void handleDoubt(Doubt doubt) {
 
     private void clearZombie(HostEndpoint zombieEndpoint) {
         // drop zombie if any, and broadcast a quit on behalf of it
-        drop(zombieEndpoint, Integer.MAX_VALUE);
+        drop(zombieEndpoint, Integer.MAX_VALUE, false);
         messenger.spread(ClusterMessage.newBuilder()
             .setQuit(Quit.newBuilder().setEndpoint(zombieEndpoint).setIncarnation(Integer.MAX_VALUE).build())
             .build());