MaintenanceTrafficResilienceIT.java

package redis.clients.jedis.scenario;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
import static org.junit.jupiter.api.Assumptions.assumeTrue;

import java.util.stream.Stream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.redis.test.fi.Scenario;
import com.redis.test.fi.StandaloneEffect;

import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.MethodSource;

import redis.clients.jedis.MaintenanceNotificationsConfig;
import redis.clients.jedis.RedisClient;

/**
 * Traffic-continuity resiliency: with a default-configured {@link RedisClient} (RESP3
 * auto-negotiated, SCH in its default AUTO mode), a topology change must be hitless ��� continuous
 * traffic before, during, and after the fault-injector effect completes sees zero command or
 * connectivity errors. This is a black-box guarantee (no notification/timeout assertions),
 * complementary to the functional suite.
 * <p>
 * Runs each MOVING-coordinated or hitless topology change the fault injector can drive on a single
 * standalone db: conn_drop and data_movement_conn_drop endpoint rebinds, and the hitless
 * data_movement_no_conn_drop migrate and failover. Rolling upgrades are out of scope (multi-node
 * orchestration, not a single standalone effect), and dns_resolution_change is excluded ��� it emits
 * no notifications, so connections drop abruptly with no handoff to coordinate and recover only via
 * DNS re-resolution.
 */
@Tag("scenario")
public class MaintenanceTrafficResilienceIT extends MaintNotificationsScenarioBase {

  private static final Logger logger = LoggerFactory
      .getLogger(MaintenanceTrafficResilienceIT.class);

  private static final int WORKERS = 6;
  private static final long BASELINE_COMMANDS = 200;
  // Keep traffic flowing this long after the effect completes to cover the MOVING grace tail ��� the
  // client-side grace timer runs past the server-reported completion ��� then drain and assert.
  private static final long POST_EFFECT_TRAFFIC_MS = 20_000;
  /** The default maintenance-notifications config (mode AUTO) the hitless guarantee is made for. */
  private static final MaintenanceNotificationsConfig DEFAULT_MAINTENANCE = MaintenanceNotificationsConfig
      .builder().build();

  static Stream<Scenario> scenarios() {
    return Stream.of(
      STANDALONE_CATALOG.effect(StandaloneEffect.CONN_DROP).trigger("endpoint_rebind").scenario(),
      STANDALONE_CATALOG.effect(StandaloneEffect.DATA_MOVEMENT_CONN_DROP).trigger("endpoint_rebind")
          .scenario(),
      STANDALONE_CATALOG.effect(StandaloneEffect.DATA_MOVEMENT_NO_CONN_DROP).trigger("migrate")
          .scenario(),
      STANDALONE_CATALOG.effect(StandaloneEffect.DATA_MOVEMENT_NO_CONN_DROP).trigger("failover")
          .scenario());
  }

  @ParameterizedTest(name = "{0}")
  @MethodSource("scenarios")
  @Timeout(420)
  void trafficUninterruptedDuringMaintenance(Scenario scenario) {
    TrafficRunner traffic = runTrafficAcrossEffect(scenario, DEFAULT_MAINTENANCE);
    if (traffic.errorCount() != 0) {
      fail("topology change must be hitless ��� " + traffic.summary(), traffic.firstError());
    }
    assertEquals("PONG", client.ping(), "client must be healthy after the effect");
  }

  /**
   * Negative control (meta-test, off by default): a drop scenario with maintenance notifications
   * DISABLED must NOT be hitless ��� the endpoint rebind drops in-flight commands with no handoff to
   * coordinate. It guards the positive test from passing vacuously ��� its hitless result must be
   * attributable to the maintenance-notifications feature. Run with
   * {@code -Dsch.negativeControl=true} (forward it to the failsafe fork, e.g. via
   * {@code -DJVM_OPTS=-Dsch.negativeControl=true}).
   */
  @Test
  @Timeout(420)
  void trafficDisruptedWithoutMaintenance() {
    assumeTrue(Boolean.getBoolean("sch.negativeControl"), "run with -Dsch.negativeControl=true");
    Scenario scenario = STANDALONE_CATALOG.effect(StandaloneEffect.CONN_DROP)
        .trigger("endpoint_rebind").scenario();
    TrafficRunner traffic = runTrafficAcrossEffect(scenario,
      MaintenanceNotificationsConfig.DISABLED);
    assertTrue(traffic.errorCount() > 0,
      "negative control: expected errors with maintenance notifications DISABLED, got none ��� "
          + traffic.summary());
  }

  /**
   * Deploys a fresh db, drives traffic before/during/after the effect with a default client (or the
   * given maintenance config), then stops and drains. Returns the runner for the caller to assert.
   */
  private TrafficRunner runTrafficAcrossEffect(Scenario scenario,
      MaintenanceNotificationsConfig maint) {
    openClient(deployDatabase(scenario), maint);

    TrafficRunner traffic = new TrafficRunner(client, WORKERS);
    traffic.start();
    try {
      // before: a healthy baseline must be error-free (no effect yet, regardless of config)
      traffic.awaitSubmitted(BASELINE_COMMANDS);
      assertEquals(0, traffic.errorCount(), "errors before the effect: " + traffic.summary());

      // during: run the effect to completion while traffic keeps flowing
      long submittedBeforeEffect = traffic.submittedCount();
      faultInjector.triggerEffect(database, scenario);

      // after: keep traffic flowing through the handoff grace tail
      long submittedAtCompletion = traffic.submittedCount();
      sleepQuietly(POST_EFFECT_TRAFFIC_MS);
      assertTrue(traffic.submittedCount() > submittedAtCompletion,
        "no traffic after the effect completed");

      logger.info("{} (maint={}): commands submitted before/at-completion/final = {}/{}/{}",
        scenario, maint.getMode(), submittedBeforeEffect, submittedAtCompletion,
        traffic.submittedCount());
    } finally {
      // stop submitting new commands and wait up to the grace timeout for in-flight ones to finish
      traffic.stop();
    }

    // confirm every submitted command reached a terminal state ��� none left in-flight
    traffic.awaitAllProcessed();
    logger.info("{} (maint={}) result: {}", scenario, maint.getMode(), traffic.summary());
    return traffic;
  }
}