Handle DNS SERVFAIL differently than NXDOMAIN

servicetalk-dns-discovery-netty/src/main/java/io/servicetalk/dns/discovery/netty/DefaultDnsClient.java

@@ -80,6 +80,7 @@
 
 import static io.netty.handler.codec.dns.DefaultDnsRecordDecoder.decodeName;
 import static io.netty.handler.codec.dns.DnsRecordType.SRV;
+import static io.netty.handler.codec.dns.DnsResponseCode.NXDOMAIN;
 import static io.servicetalk.client.api.ServiceDiscovererEvent.Status.AVAILABLE;
 import static io.servicetalk.concurrent.api.AsyncCloseables.toAsyncCloseable;
 import static io.servicetalk.concurrent.api.Completable.completed;
@@ -132,18 +133,18 @@ final class DefaultDnsClient implements DnsClient {
     @Nullable
     private final DnsServiceDiscovererObserver observer;
     private final ServiceDiscovererEvent.Status missingRecordStatus;
+    private final boolean nxInvalidation;
     private final IntFunction<? extends Completable> srvHostNameRepeater;
     private final int srvConcurrency;
     private final boolean srvFilterDuplicateEvents;
-    private final boolean inactiveEventsOnError;
     private final DnsResolverAddressTypes addressTypes;
     private final String id;
     private boolean closed;
 
     DefaultDnsClient(final String id, final IoExecutor ioExecutor, final int consolidateCacheSize,
                      final int minTTL, final int maxTTL, final int minCacheTTL, final int maxCacheTTL,
                      final int negativeTTLCacheSeconds, final long ttlJitterNanos,
-                     final int srvConcurrency, final boolean inactiveEventsOnError,
+                     final int srvConcurrency,
                      final boolean completeOncePreferredResolved, final boolean srvFilterDuplicateEvents,
                      Duration srvHostNameRepeatInitialDelay, Duration srvHostNameRepeatJitter,
                      @Nullable Integer maxUdpPayloadSize, @Nullable final Integer ndots,
@@ -152,10 +153,10 @@ final class DefaultDnsClient implements DnsClient {
                      @Nullable final SocketAddress localAddress,
                      @Nullable final DnsServerAddressStreamProvider dnsServerAddressStreamProvider,
                      @Nullable final DnsServiceDiscovererObserver observer,
-                     final ServiceDiscovererEvent.Status missingRecordStatus) {
+                     final ServiceDiscovererEvent.Status missingRecordStatus,
+                     final boolean nxInvalidation) {
         this.srvConcurrency = srvConcurrency;
         this.srvFilterDuplicateEvents = srvFilterDuplicateEvents;
-        this.inactiveEventsOnError = inactiveEventsOnError;
         // Implementation of this class expects to use only single EventLoop from IoExecutor
         this.nettyIoExecutor = toEventLoopAwareNettyIoExecutor(ioExecutor).next();
         // We must use nettyIoExecutor for the repeater for thread safety!
@@ -170,6 +171,7 @@ final class DefaultDnsClient implements DnsClient {
         this.addressTypes = dnsResolverAddressTypes;
         this.observer = observer;
         this.missingRecordStatus = missingRecordStatus;
+        this.nxInvalidation = nxInvalidation;
         this.id = id + '@' + toHexString(identityHashCode(this));
         asyncCloseable = toAsyncCloseable(graceful -> {
             if (nettyIoExecutor.isCurrentThreadEventLoop()) {
@@ -252,9 +254,8 @@ public Publisher<Collection<ServiceDiscovererEvent<InetAddress>>> dnsQuery(final
         return defer(() -> {
             final DnsDiscoveryObserver discoveryObserver = newDiscoveryObserver(address);
             ARecordPublisher pub = new ARecordPublisher(address, discoveryObserver);
-            Publisher<? extends Collection<ServiceDiscovererEvent<InetAddress>>> events = inactiveEventsOnError ?
-                    recoverWithInactiveEvents(pub, false) :
-                    pub;
+            Publisher<? extends Collection<ServiceDiscovererEvent<InetAddress>>> events =
+                    recoverWithInactiveEvents(pub, false, nxInvalidation);
             return discoveryObserver == null ? events : events.beforeFinally(new TerminalSignalConsumer() {
                     @Override
                     public void onComplete() {
@@ -298,7 +299,7 @@ public Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQu
         // any pending scheduled tasks. SrvInactiveCombinerOperator is used to filter the aggregated collection of
         // inactive events if necessary.
         Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> events =
-                recoverWithInactiveEvents(new SrvRecordPublisher(serviceName, discoveryObserver), true)
+                recoverWithInactiveEvents(new SrvRecordPublisher(serviceName, discoveryObserver), true, nxInvalidation)
                 .flatMapConcatIterable(identity())
                 .flatMapMerge(srvEvent -> {
                 assertInEventloop();
@@ -312,8 +313,10 @@ public Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQu
                             return newDuplicateSrv(serviceName, srvEvent.address().hostName());
                         }
 
+                        // NXDOMAIN = invalidation for A queries part of SRV lookups for backwards compatibility.
+                        // This is a behavior difference between plain A lookups and SRV rooted A lookups.
                         Publisher<? extends Collection<ServiceDiscovererEvent<InetAddress>>> returnPub =
-                                recoverWithInactiveEvents(aPublisher, false);
+                                recoverWithInactiveEvents(aPublisher, false, true);
                         return srvFilterDuplicateEvents ?
                                 srvFilterDups(returnPub, availableAddresses, srvEvent.address().port()) :
                                 returnPub.map(ev -> mapEventList(ev, inetAddress ->
@@ -338,7 +341,7 @@ public Publisher<Collection<ServiceDiscovererEvent<InetSocketAddress>>> dnsSrvQu
                     return empty();
                 }
             }, srvConcurrency)
-            .liftSync(inactiveEventsOnError ? SrvInactiveCombinerOperator.EMIT : SrvInactiveCombinerOperator.NO_EMIT);
+            .liftSync(SrvInactiveCombinerOperator.EMIT);
 
             return discoveryObserver == null ? events : events.beforeFinally(new TerminalSignalConsumer() {
                 @Override
@@ -952,10 +955,10 @@ private static Publisher<? extends Collection<ServiceDiscovererEvent<InetSocketA
     }
 
     private static <T, A> Publisher<? extends Collection<ServiceDiscovererEvent<T>>> recoverWithInactiveEvents(
-            AbstractDnsPublisher<T> pub, boolean generateAggregateEvent) {
+            AbstractDnsPublisher<T> pub, boolean generateAggregateEvent, boolean nxInvalidation) {
         return pub.onErrorResume(cause -> {
             AbstractDnsPublisher<T>.AbstractDnsSubscription subscription = pub.subscription;
-            if (subscription != null) {
+            if (subscription != null && shouldRevokeState(cause, nxInvalidation)) {
                 List<ServiceDiscovererEvent<T>> events = subscription.generateInactiveEvent();
                 if (!events.isEmpty()) {
                     return (generateAggregateEvent ? Publisher.<List<ServiceDiscovererEvent<T>>>from(
@@ -968,6 +971,16 @@ private static <T, A> Publisher<? extends Collection<ServiceDiscovererEvent<T>>>
         });
     }
 
+    private static boolean shouldRevokeState(final Throwable t, final boolean nxInvalidation) {
+        // ISE => Subscriber exceptions (downstream of retry)
+        return t instanceof SrvAddressRemovedException || t instanceof IllegalStateException ||
+                t instanceof ClosedDnsServiceDiscovererException || (nxInvalidation &&
+                // string matching is done on purpose to avoid the hard Netty dependency
+                (t.getCause() != null && t.getCause().getClass().getName()
+                        .equals("io.netty.resolver.dns.DnsErrorCauseException")) &&
+                NXDOMAIN.equals(((io.netty.resolver.dns.DnsErrorCauseException) t.getCause()).getCode()));
+    }
+
     private static <T> Publisher<T> newDuplicateSrv(String serviceName, String resolvedAddress) {
         return failed(new IllegalStateException("Duplicate SRV entry for SRV name " + serviceName + " for address " +
                 resolvedAddress));
@@ -982,7 +995,6 @@ private static final class SrvInactiveCombinerOperator implements
                             PublisherOperator<Collection<ServiceDiscovererEvent<InetSocketAddress>>,
                                               Collection<ServiceDiscovererEvent<InetSocketAddress>>> {
         static final SrvInactiveCombinerOperator EMIT = new SrvInactiveCombinerOperator(true);
-        static final SrvInactiveCombinerOperator NO_EMIT = new SrvInactiveCombinerOperator(false);
         private final boolean emitAggregatedEvents;
 
         private SrvInactiveCombinerOperator(boolean emitAggregatedEvents) {

servicetalk-dns-discovery-netty/src/main/java/io/servicetalk/dns/discovery/netty/DefaultDnsServiceDiscovererBuilder.java

@@ -62,6 +62,8 @@ public final class DefaultDnsServiceDiscovererBuilder implements DnsServiceDisco
      */
     @Deprecated // FIXME: 0.43 - consider removing this system property
     private static final String SKIP_BINDING_PROPERTY = "io.servicetalk.dns.discovery.netty.skipBinding";
+    private static final String NX_DOMAIN_INVALIDATES_PROPERTY = "io.servicetalk.dns.discovery.nxdomain.invalidation";
+    static final boolean DEFAULT_NX_DOMAIN_INVALIDATES = parseProperty(NX_DOMAIN_INVALIDATES_PROPERTY, false);
     @Nullable
     private static final SocketAddress DEFAULT_LOCAL_ADDRESS =
             getBoolean(SKIP_BINDING_PROPERTY) ? null : new InetSocketAddress(0);
@@ -101,6 +103,7 @@ public final class DefaultDnsServiceDiscovererBuilder implements DnsServiceDisco
             LOGGER.debug("-D{}={}", NEGATIVE_TTL_CACHE_SECONDS_PROPERTY, negativeCacheTtlValue);
             LOGGER.debug("Default negative TTL cache in seconds: {}", DEFAULT_NEGATIVE_TTL_CACHE_SECONDS);
             LOGGER.debug("Default missing records status: {}", DEFAULT_MISSING_RECOREDS_STATUS);
+            LOGGER.debug("-D{}: {}", NX_DOMAIN_INVALIDATES_PROPERTY, DEFAULT_NX_DOMAIN_INVALIDATES);
         }
     }
 
@@ -128,7 +131,6 @@ public final class DefaultDnsServiceDiscovererBuilder implements DnsServiceDisco
     private int negativeTTLCacheSeconds = DEFAULT_NEGATIVE_TTL_CACHE_SECONDS;
     private Duration ttlJitter = ofSeconds(DEFAULT_TTL_POLL_JITTER_SECONDS);
     private int srvConcurrency = 2048;
-    private boolean inactiveEventsOnError;
     private boolean completeOncePreferredResolved = true;
     private boolean srvFilterDuplicateEvents;
     private Duration srvHostNameRepeatInitialDelay = ofSeconds(10);
@@ -138,6 +140,7 @@ public final class DefaultDnsServiceDiscovererBuilder implements DnsServiceDisco
     @Nullable
     private DnsServiceDiscovererObserver observer;
     private ServiceDiscovererEvent.Status missingRecordStatus = DEFAULT_MISSING_RECOREDS_STATUS;
+    private boolean nxInvalidation = DEFAULT_NX_DOMAIN_INVALIDATES;
 
     /**
      * Creates a new {@link DefaultDnsServiceDiscovererBuilder}.
@@ -288,6 +291,18 @@ public DefaultDnsServiceDiscovererBuilder missingRecordStatus(ServiceDiscovererE
         return this;
     }
 
+    /**
+     * Modify the behavior of the system flag about invalidating DNS state when NXDOMAIN is seen.
+     * Default behavior is controlled through {@link #NX_DOMAIN_INVALIDATES_PROPERTY}.
+     *
+     * @param nxInvalidation Flag to enable/disable behavior.
+     * @return {@code this} builder.
+     */
+    DefaultDnsServiceDiscovererBuilder nxInvalidates(final boolean nxInvalidation) {
+        this.nxInvalidation = nxInvalidation;
+        return this;
+    }
+
     @Override
     public ServiceDiscoverer<String, InetSocketAddress, ServiceDiscovererEvent<InetSocketAddress>>
     buildSrvDiscoverer() {
@@ -300,11 +315,6 @@ public DefaultDnsServiceDiscovererBuilder missingRecordStatus(ServiceDiscovererE
         return asHostAndPortDiscoverer(build());
     }
 
-    DefaultDnsServiceDiscovererBuilder inactiveEventsOnError(boolean inactiveEventsOnError) {
-        this.inactiveEventsOnError = inactiveEventsOnError;
-        return this;
-    }
-
     DefaultDnsServiceDiscovererBuilder srvConcurrency(int srvConcurrency) {
         this.srvConcurrency = ensurePositive(srvConcurrency, "srvConcurrency");
         return this;
@@ -317,8 +327,11 @@ DefaultDnsServiceDiscovererBuilder completeOncePreferredResolved(boolean complet
 
     DefaultDnsServiceDiscovererBuilder srvHostNameRepeatDelay(
             Duration initialDelay, Duration jitter) {
-        this.srvHostNameRepeatInitialDelay = requireNonNull(initialDelay);
-        this.srvHostNameRepeatJitter = requireNonNull(jitter);
+        this.srvHostNameRepeatInitialDelay = ensurePositive(initialDelay, "srvHostNameRepeatInitialDelay");
+        this.srvHostNameRepeatJitter = ensurePositive(jitter, "srvHostNameRepeatJitter");
+        if (srvHostNameRepeatJitter.toNanos() >= srvHostNameRepeatInitialDelay.toNanos()) {
+            throw new IllegalArgumentException("The jitter value should be less than the initial delay.");
+        }
         return this;
     }
 
@@ -369,10 +382,10 @@ DnsClient build() {
                 ioExecutor == null ? globalExecutionContext().ioExecutor() : ioExecutor, consolidateCacheSize,
                 minTTLSeconds, maxTTLSeconds, minTTLCacheSeconds, maxTTLCacheSeconds, negativeTTLCacheSeconds,
                 ttlJitter.toNanos(),
-                srvConcurrency, inactiveEventsOnError, completeOncePreferredResolved, srvFilterDuplicateEvents,
+                srvConcurrency, completeOncePreferredResolved, srvFilterDuplicateEvents,
                 srvHostNameRepeatInitialDelay, srvHostNameRepeatJitter, maxUdpPayloadSize, ndots, optResourceEnabled,
                 queryTimeout, dnsResolverAddressTypes, localAddress, dnsServerAddressStreamProvider, observer,
-                missingRecordStatus);
+                missingRecordStatus, nxInvalidation);
         return filterFactory == null ? rawClient : filterFactory.create(rawClient);
     }
 
@@ -389,4 +402,12 @@ private static int parseProperty(final String name, final int defaultValue) {
             return defaultValue;
         }
     }
+
+    private static boolean parseProperty(final String name, final boolean defaultValue) {
+        final String value = getProperty(name);
+        if (value == null) {
+            return defaultValue;
+        }
+        return Boolean.parseBoolean(value);
+    }
 }