From aa6d0b29b1554feaa47907ad4d95afb5ca9afcf5 Mon Sep 17 00:00:00 2001 From: Hellblazer Date: Wed, 12 Jun 2024 08:41:14 -0700 Subject: [PATCH] tighten up the screws --- .../apollo/choam/GenesisAssembly.java | 17 ++++---- .../salesforce/apollo/choam/Parameters.java | 22 +--------- .../com/salesforce/apollo/choam/Producer.java | 3 +- .../salesforce/apollo/choam/ViewAssembly.java | 1 + .../salesforce/apollo/choam/fsm/Genesis.java | 11 ++++- .../apollo/choam/fsm/Reconfiguration.java | 5 +++ .../apollo/choam/support/TxDataSource.java | 8 ++-- .../salesforce/apollo/choam/DynamicTest.java | 6 +-- .../apollo/choam/MembershipTests.java | 1 - .../choam/support/TxDataSourceTest.java | 17 ++++---- .../cryptography/SignatureAlgorithm.java | 2 +- .../com/salesforce/apollo/fireflies/View.java | 2 +- .../apollo/model/ContainmentDomainTest.java | 42 ++++++++----------- .../salesforce/apollo/model/DomainTest.java | 6 +-- .../apollo/model/FireFliesTest.java | 32 +++++++------- .../apollo/state/AbstractLifecycleTest.java | 1 - 16 files changed, 73 insertions(+), 103 deletions(-) diff --git a/choam/src/main/java/com/salesforce/apollo/choam/GenesisAssembly.java b/choam/src/main/java/com/salesforce/apollo/choam/GenesisAssembly.java index a5ee566f6..3b68bc7a8 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/GenesisAssembly.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/GenesisAssembly.java @@ -93,7 +93,7 @@ public GenesisAssembly(ViewContext vc, CommonCommunications comms, } else { config.setPid(pid).setnProc((short) view.roster().size()); } - config.setEpochLength(7).setNumberOfEpochs(3); + config.setEpochLength(33).setNumberOfEpochs(-1); config.setLabel("Genesis Assembly" + view.context().getId() + " on: " + params().member().getId()); controller = new Ethereal(config.build(), params().producer().maxBatchByteSize(), dataSource(), transitions::process, transitions::nextEpoch, label); @@ -106,13 +106,11 @@ public GenesisAssembly(ViewContext vc, CommonCommunications comms, @Override public void certify() { - if (slate.size() < params().majority()) { - log.info("Not certifying genesis for: {} need: {} slate incomplete: {} on: {}", view.context().getId(), - params().majority(), slate.keySet().stream().sorted().toList(), params().member().getId()); + if (slate.size() != nextAssembly.size()) { + log.info("Not certifying genesis for: {} slate incomplete: {} on: {}", view.context().getId(), + slate.keySet().stream().sorted().toList(), params().member().getId()); return; } - assert slate.size() >= params().majority() : "Expected: %s members, slate: %s".formatted(params().majority(), - slate.size()); reconfiguration = new HashedBlock(params().digestAlgorithm(), view.genesis(slate, view.context().getId(), new NullBlock( params().digestAlgorithm()))); @@ -163,6 +161,9 @@ public void gather(List preblock, boolean last) { .peek(j -> log.info("Gathering: {} on: {}", Digest.from(j.getMember().getVm().getId()), params().member().getId())) .forEach(this::join); + if (slate.size() == nextAssembly.size()) { + transitions.gathered(); + } } @Override @@ -187,12 +188,12 @@ public void publish() { log.trace("Cannot publish genesis, reconfiguration is NULL on: {}", params().member().getId()); return; } - if (witnesses.size() < params().majority()) { + if (witnesses.size() < nextAssembly.size()) { log.trace("Cannot publish genesis: {} with: {} witnesses on: {}", reconfiguration.hash, witnesses.size(), params().member().getId()); return; } - if (reconfiguration.block.getGenesis().getInitialView().getJoinsCount() < params().majority()) { + if (reconfiguration.block.getGenesis().getInitialView().getJoinsCount() < nextAssembly.size()) { log.trace("Cannot publish genesis: {} with: {} joins on: {}", reconfiguration.hash, reconfiguration.block.getGenesis().getInitialView().getJoinsCount(), params().member().getId()); return; diff --git a/choam/src/main/java/com/salesforce/apollo/choam/Parameters.java b/choam/src/main/java/com/salesforce/apollo/choam/Parameters.java index 2a3449d64..6241fb072 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/Parameters.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/Parameters.java @@ -56,7 +56,7 @@ public record Parameters(Parameters.RuntimeParameters runtime, ReliableBroadcast Parameters.BootstrapParameters bootstrap, Parameters.ProducerParameters producer, Parameters.MvStoreBuilder mvBuilder, Parameters.LimiterBuilder txnLimiterBuilder, ExponentialBackoffPolicy.Builder submitPolicy, int checkpointSegmentSize, - ExponentialBackoffPolicy.Builder drainPolicy, boolean generateGenesis) { + boolean generateGenesis) { public static Builder newBuilder() { return new Builder(); @@ -677,14 +677,6 @@ public static class Builder implements Cloneable { private ReliableBroadcaster.Parameters combine = ReliableBroadcaster.Parameters.newBuilder() .build(); private DigestAlgorithm digestAlgorithm = DigestAlgorithm.DEFAULT; - private ExponentialBackoffPolicy.Builder drainPolicy = ExponentialBackoffPolicy.newBuilder() - .setInitialBackoff( - Duration.ofMillis(5)) - .setJitter(0.2) - .setMultiplier(1.2) - .setMaxBackoff( - Duration.ofMillis( - 500)); private Digest genesisViewId; private Duration gossipDuration = Duration.ofSeconds(1); private int maxCheckpointSegments = 200; @@ -709,7 +701,7 @@ public Parameters build(RuntimeParameters runtime) { return new Parameters(runtime, combine, gossipDuration, maxCheckpointSegments, submitTimeout, genesisViewId, checkpointBlockDelta, crowns, digestAlgorithm, viewSigAlgorithm, synchronizationCycles, regenerationCycles, bootstrap, producer, mvBuilder, - txnLimiterBuilder, submitPolicy, checkpointSegmentSize, drainPolicy, generateGenesis); + txnLimiterBuilder, submitPolicy, checkpointSegmentSize, generateGenesis); } @Override @@ -726,7 +718,6 @@ public Builder clone() { producer.batchInterval, producer.maxBatchCount(), producer.maxGossipDelay)); clone.setTxnLimiterBuilder(txnLimiterBuilder.clone()); clone.setSubmitPolicy(submitPolicy.clone()); - clone.setDrainPolicy(drainPolicy.clone()); return clone; } @@ -783,15 +774,6 @@ public Builder setDigestAlgorithm(DigestAlgorithm digestAlgorithm) { return this; } - public ExponentialBackoffPolicy.Builder getDrainPolicy() { - return drainPolicy; - } - - public Builder setDrainPolicy(ExponentialBackoffPolicy.Builder drainPolicy) { - this.drainPolicy = drainPolicy; - return this; - } - public Digest getGenesisViewId() { return genesisViewId; } diff --git a/choam/src/main/java/com/salesforce/apollo/choam/Producer.java b/choam/src/main/java/com/salesforce/apollo/choam/Producer.java index 19bc5580f..1aae471cb 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/Producer.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/Producer.java @@ -76,8 +76,7 @@ public Producer(Digest nextViewId, ViewContext view, HashedBlock lastBlock, Hash maxEpoch = ep.getEpochLength(); ds = new TxDataSource(params.member(), blocks, params.metrics(), producerParams.maxBatchByteSize(), - producerParams.batchInterval(), producerParams.maxBatchCount(), - params().drainPolicy().build()); + producerParams.batchInterval(), producerParams.maxBatchCount()); log.debug("Producer max elements: {} reconfiguration epoch: {} on: {}", blocks, maxEpoch, params.member().getId()); diff --git a/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java b/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java index a1503ce9a..cf31987be 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/ViewAssembly.java @@ -401,6 +401,7 @@ public String toString() { private class Recon implements Reconfiguration { @Override public void certify() { + countdown.set(-1); if (proposals.size() == selected.assembly.size()) { log.info("Certifying: {} majority: {} of: {} slate: {} on: {}", nextViewId, selected.majority, nextViewId, proposals.keySet().stream().sorted().toList(), params().member().getId()); diff --git a/choam/src/main/java/com/salesforce/apollo/choam/fsm/Genesis.java b/choam/src/main/java/com/salesforce/apollo/choam/fsm/Genesis.java index e917bc716..e2b5a2532 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/fsm/Genesis.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/fsm/Genesis.java @@ -55,8 +55,12 @@ public void gather() { @Override public Transitions nextEpoch(Integer epoch) { - return epoch.equals(0) ? null : CERTIFICATION; + return null; + } + @Override + public Transitions gathered() { + return CERTIFICATION; } @Override @@ -80,11 +84,14 @@ public Transitions process(List preblock, boolean last) { return null; } } - } interface Transitions extends FsmExecutor { + default Transitions gathered() { + throw fsm().invalidTransitionOn(); + } + default Transitions nextEpoch(Integer epoch) { throw fsm().invalidTransitionOn(); } diff --git a/choam/src/main/java/com/salesforce/apollo/choam/fsm/Reconfiguration.java b/choam/src/main/java/com/salesforce/apollo/choam/fsm/Reconfiguration.java index 174cac3ac..8e9dc9a49 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/fsm/Reconfiguration.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/fsm/Reconfiguration.java @@ -75,6 +75,11 @@ public Transitions countdownCompleted() { return RECONFIGURED; } + @Override + public Transitions checkAssembly() { + return null; + } + // See if we already have a full complement of Joins of the next committee // if not set a deadline @Entry diff --git a/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java b/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java index 8053d4878..de81a7712 100644 --- a/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java +++ b/choam/src/main/java/com/salesforce/apollo/choam/support/TxDataSource.java @@ -38,7 +38,6 @@ public class TxDataSource implements DataSource { private final Duration batchInterval; private final AtomicBoolean draining = new AtomicBoolean(); - private final ExponentialBackoffPolicy drainPolicy; private final Member member; private final ChoamMetrics metrics; private final BatchingQueue processing; @@ -47,10 +46,9 @@ public class TxDataSource implements DataSource { private volatile Thread blockingThread; public TxDataSource(Member member, int maxElements, ChoamMetrics metrics, int maxBatchByteSize, - Duration batchInterval, int maxBatchCount, ExponentialBackoffPolicy drainPolicy) { + Duration batchInterval, int maxBatchCount) { this.member = member; this.batchInterval = batchInterval; - this.drainPolicy = drainPolicy; processing = new BatchingQueue(maxElements, maxBatchCount, tx -> tx.toByteString().size(), maxBatchByteSize); this.metrics = metrics; @@ -85,7 +83,7 @@ public ByteString getData() { var v = new ArrayList(); if (draining.get()) { - var target = Instant.now().plus(drainPolicy.nextBackoff()); + var target = Instant.now().plus(batchInterval); while (target.isAfter(Instant.now()) && builder.getAssembliesCount() == 0 && builder.getValidationsCount() == 0) { // rinse and repeat @@ -103,7 +101,7 @@ public ByteString getData() { // sleep waiting for input try { - Thread.sleep(drainPolicy.getInitialBackoff().dividedBy(2).toMillis()); + Thread.sleep(batchInterval.dividedBy(2).toMillis()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return ByteString.EMPTY; diff --git a/choam/src/test/java/com/salesforce/apollo/choam/DynamicTest.java b/choam/src/test/java/com/salesforce/apollo/choam/DynamicTest.java index 7cb99d6d1..332da3e8b 100644 --- a/choam/src/test/java/com/salesforce/apollo/choam/DynamicTest.java +++ b/choam/src/test/java/com/salesforce/apollo/choam/DynamicTest.java @@ -4,7 +4,6 @@ import com.salesforce.apollo.archipelago.Router; import com.salesforce.apollo.archipelago.ServerConnectionCache; import com.salesforce.apollo.archipelago.UnsafeExecutors; -import com.salesforce.apollo.choam.support.ExponentialBackoffPolicy; import com.salesforce.apollo.context.Context; import com.salesforce.apollo.context.DynamicContext; import com.salesforce.apollo.cryptography.DigestAlgorithm; @@ -86,10 +85,7 @@ public void setUp() throws Exception { .setNumberOfEpochs(3) .setEpochLength(7)) .build()) - .setCheckpointBlockDelta(checkpointBlockSize) - .setDrainPolicy(ExponentialBackoffPolicy.newBuilder() - .setInitialBackoff(Duration.ofMillis(1)) - .setMaxBackoff(Duration.ofMillis(1))); + .setCheckpointBlockDelta(checkpointBlockSize); members.subList(0, 4).forEach(m -> { var context = (DynamicContext) contextBuilder.build(); diff --git a/choam/src/test/java/com/salesforce/apollo/choam/MembershipTests.java b/choam/src/test/java/com/salesforce/apollo/choam/MembershipTests.java index effd854f5..f9324e628 100644 --- a/choam/src/test/java/com/salesforce/apollo/choam/MembershipTests.java +++ b/choam/src/test/java/com/salesforce/apollo/choam/MembershipTests.java @@ -152,7 +152,6 @@ public SigningMember initialize(int checkpointBlockSize, int cardinality) throws .build()) .setGenerateGenesis(true) .setCheckpointBlockDelta(checkpointBlockSize); - params.getDrainPolicy().setInitialBackoff(Duration.ofMillis(1)).setMaxBackoff(Duration.ofMillis(1)); params.getProducer().ethereal().setNumberOfEpochs(2).setEpochLength(20); var entropy = SecureRandom.getInstance("SHA1PRNG"); diff --git a/choam/src/test/java/com/salesforce/apollo/choam/support/TxDataSourceTest.java b/choam/src/test/java/com/salesforce/apollo/choam/support/TxDataSourceTest.java index 4f89b2f7f..2583a2bc9 100644 --- a/choam/src/test/java/com/salesforce/apollo/choam/support/TxDataSourceTest.java +++ b/choam/src/test/java/com/salesforce/apollo/choam/support/TxDataSourceTest.java @@ -6,15 +6,6 @@ */ package com.salesforce.apollo.choam.support; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNotNull; - -import java.security.SecureRandom; -import java.time.Duration; - -import org.junit.jupiter.api.Test; - import com.google.protobuf.ByteString; import com.salesforce.apollo.choam.proto.Transaction; import com.salesforce.apollo.cryptography.DigestAlgorithm; @@ -22,6 +13,12 @@ import com.salesforce.apollo.stereotomy.StereotomyImpl; import com.salesforce.apollo.stereotomy.mem.MemKERL; import com.salesforce.apollo.stereotomy.mem.MemKeyStore; +import org.junit.jupiter.api.Test; + +import java.security.SecureRandom; +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.*; /** * @author hal.hildebrand @@ -34,7 +31,7 @@ public void func() throws Exception { entropy.setSeed(new byte[] { 6, 6, 6 }); var stereotomy = new StereotomyImpl(new MemKeyStore(), new MemKERL(DigestAlgorithm.DEFAULT), entropy); TxDataSource ds = new TxDataSource(new ControlledIdentifierMember(stereotomy.newIdentifier()), 100, null, 1024, - Duration.ofMillis(100), 100, ExponentialBackoffPolicy.newBuilder().build()); + Duration.ofMillis(100), 100); Transaction tx = Transaction.newBuilder() .setContent(ByteString.copyFromUtf8("Give me food or give me slack or kill me")) .build(); diff --git a/cryptography/src/main/java/com/salesforce/apollo/cryptography/SignatureAlgorithm.java b/cryptography/src/main/java/com/salesforce/apollo/cryptography/SignatureAlgorithm.java index 1e6f868a0..c69097ecb 100644 --- a/cryptography/src/main/java/com/salesforce/apollo/cryptography/SignatureAlgorithm.java +++ b/cryptography/src/main/java/com/salesforce/apollo/cryptography/SignatureAlgorithm.java @@ -251,7 +251,7 @@ JohnHancock sign(ULong sequenceNumber, PrivateKey[] privateKeys, InputStream mes public static SignatureAlgorithm fromSignatureCode(int i) { return switch (i) { case 0: - throw new IllegalArgumentException("Unknown signature code: " + i); + yield NULL_SIGNATURE; case 1: yield NULL_SIGNATURE; case 2: diff --git a/fireflies/src/main/java/com/salesforce/apollo/fireflies/View.java b/fireflies/src/main/java/com/salesforce/apollo/fireflies/View.java index 139d025b6..15d93295d 100644 --- a/fireflies/src/main/java/com/salesforce/apollo/fireflies/View.java +++ b/fireflies/src/main/java/com/salesforce/apollo/fireflies/View.java @@ -1858,7 +1858,7 @@ public void join(Join join, Digest from, StreamObserver responseObserve /** * The first message in the anti-entropy protocol. Process any digests from the inbound gossip digest. Respond * with the Gossip that represents the digests newer or not known in this view, as well as updates from this - * node based on out of date information in the supplied digests. + * node based on out-of-date information in the supplied digests. * * @param request - the Gossip from our partner * @return Teh response for Moar gossip - updates this node has which the sender is out of touch with, and diff --git a/model/src/test/java/com/salesforce/apollo/model/ContainmentDomainTest.java b/model/src/test/java/com/salesforce/apollo/model/ContainmentDomainTest.java index 641c211c8..e8b6982c2 100644 --- a/model/src/test/java/com/salesforce/apollo/model/ContainmentDomainTest.java +++ b/model/src/test/java/com/salesforce/apollo/model/ContainmentDomainTest.java @@ -11,7 +11,6 @@ import com.salesforce.apollo.choam.Parameters.Builder; import com.salesforce.apollo.choam.Parameters.RuntimeParameters; import com.salesforce.apollo.choam.proto.FoundationSeal; -import com.salesforce.apollo.choam.support.ExponentialBackoffPolicy; import com.salesforce.apollo.context.DynamicContextImpl; import com.salesforce.apollo.cryptography.Digest; import com.salesforce.apollo.cryptography.DigestAlgorithm; @@ -116,33 +115,28 @@ public void smoke() throws Exception { .filter(c -> !c.active()) .map(Domain::logState) .toList())); - var oracle = domains.get(0).getDelphi(); + var oracle = domains.getFirst().getDelphi(); oracle.add(new Oracle.Namespace("test")).get(); DomainTest.smoke(oracle); } private Builder params() { - var template = Parameters.newBuilder() - .setGenerateGenesis(true) - .setGenesisViewId(GENESIS_VIEW_ID) - .setBootstrap(Parameters.BootstrapParameters.newBuilder() - .setGossipDuration(Duration.ofMillis(20)) - .build()) - .setGenesisViewId(DigestAlgorithm.DEFAULT.getOrigin()) - .setGossipDuration(Duration.ofMillis(20)) - .setProducer(Parameters.ProducerParameters.newBuilder() - .setGossipDuration(Duration.ofMillis(20)) - .setBatchInterval(Duration.ofMillis(50)) - .setMaxBatchByteSize(1024 * 1024) - .setMaxBatchCount(10_000) - .setEthereal(Config.newBuilder() - .setNumberOfEpochs(12) - .setEpochLength(33)) - .build()) - .setCheckpointBlockDelta(200) - .setDrainPolicy(ExponentialBackoffPolicy.newBuilder() - .setInitialBackoff(Duration.ofMillis(1)) - .setMaxBackoff(Duration.ofMillis(1))); - return template; + return Parameters.newBuilder() + .setGenerateGenesis(true) + .setGenesisViewId(GENESIS_VIEW_ID) + .setBootstrap( + Parameters.BootstrapParameters.newBuilder().setGossipDuration(Duration.ofMillis(20)).build()) + .setGenesisViewId(DigestAlgorithm.DEFAULT.getOrigin()) + .setGossipDuration(Duration.ofMillis(20)) + .setProducer(Parameters.ProducerParameters.newBuilder() + .setGossipDuration(Duration.ofMillis(20)) + .setBatchInterval(Duration.ofMillis(50)) + .setMaxBatchByteSize(1024 * 1024) + .setMaxBatchCount(10_000) + .setEthereal(Config.newBuilder() + .setNumberOfEpochs(12) + .setEpochLength(33)) + .build()) + .setCheckpointBlockDelta(200); } } diff --git a/model/src/test/java/com/salesforce/apollo/model/DomainTest.java b/model/src/test/java/com/salesforce/apollo/model/DomainTest.java index 122956188..ff25f7d56 100644 --- a/model/src/test/java/com/salesforce/apollo/model/DomainTest.java +++ b/model/src/test/java/com/salesforce/apollo/model/DomainTest.java @@ -11,7 +11,6 @@ import com.salesforce.apollo.choam.Parameters.Builder; import com.salesforce.apollo.choam.Parameters.RuntimeParameters; import com.salesforce.apollo.choam.proto.FoundationSeal; -import com.salesforce.apollo.choam.support.ExponentialBackoffPolicy; import com.salesforce.apollo.context.DynamicContextImpl; import com.salesforce.apollo.cryptography.Digest; import com.salesforce.apollo.cryptography.DigestAlgorithm; @@ -297,10 +296,7 @@ private Builder params() { .setNumberOfEpochs(12) .setEpochLength(33)) .build()) - .setCheckpointBlockDelta(200) - .setDrainPolicy(ExponentialBackoffPolicy.newBuilder() - .setInitialBackoff(Duration.ofMillis(1)) - .setMaxBackoff(Duration.ofMillis(1))); + .setCheckpointBlockDelta(200); return template; } } diff --git a/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java b/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java index 867d8d4d0..0b3105f54 100644 --- a/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java +++ b/model/src/test/java/com/salesforce/apollo/model/FireFliesTest.java @@ -11,7 +11,6 @@ import com.salesforce.apollo.choam.Parameters.Builder; import com.salesforce.apollo.choam.Parameters.RuntimeParameters; import com.salesforce.apollo.choam.proto.FoundationSeal; -import com.salesforce.apollo.choam.support.ExponentialBackoffPolicy; import com.salesforce.apollo.context.Context; import com.salesforce.apollo.context.DynamicContextImpl; import com.salesforce.apollo.cryptography.Digest; @@ -57,7 +56,7 @@ public class FireFliesTest { @AfterEach public void after() { - domains.forEach(n -> n.stop()); + domains.forEach(ProcessDomain::stop); domains.clear(); routers.values().forEach(r -> r.close(Duration.ofSeconds(0))); routers.clear(); @@ -78,9 +77,10 @@ public void before() throws Exception { var params = params(); var stereotomy = new StereotomyImpl(new MemKeyStore(), new MemKERL(params.getDigestAlgorithm()), entropy); - var identities = IntStream.range(0, CARDINALITY).mapToObj(i -> { - return stereotomy.newIdentifier(); - }).collect(Collectors.toMap(controlled -> controlled.getIdentifier().getDigest(), controlled -> controlled)); + var identities = IntStream.range(0, CARDINALITY) + .mapToObj(i -> stereotomy.newIdentifier()) + .collect(Collectors.toMap(controlled -> controlled.getIdentifier().getDigest(), + controlled -> controlled)); Digest group = DigestAlgorithm.DEFAULT.getOrigin(); var sealed = FoundationSeal.newBuilder().build(); @@ -144,10 +144,10 @@ public void smokin() throws Exception { assertTrue(countdown.await(30, TimeUnit.SECONDS), "Could not join all members in all views"); - assertTrue(Utils.waitForCondition(60_000, 1_000, () -> { - return domains.stream().filter(d -> d.getFoundation().getContext().activeCount() != domains.size()).count() - == 0; - })); + assertTrue(Utils.waitForCondition(60_000, 1_000, () -> domains.stream() + .noneMatch( + d -> d.getFoundation().getContext().activeCount() + != domains.size()))); System.out.println(); System.out.println("******"); System.out.println( @@ -155,12 +155,11 @@ public void smokin() throws Exception { + " members"); System.out.println("******"); System.out.println(); - domains.forEach(n -> n.start()); - final var activated = Utils.waitForCondition(20_000, 1_000, - () -> domains.stream().filter(c -> !c.active()).count() == 0); + domains.forEach(ProcessDomain::start); + final var activated = Utils.waitForCondition(20_000, 1_000, () -> domains.stream().allMatch(Domain::active)); assertTrue(activated, "Domains did not become active : " + (domains.stream() .filter(c -> !c.active()) - .map(d -> d.logState()) + .map(Domain::logState) .toList())); System.out.println(); System.out.println("******"); @@ -169,7 +168,7 @@ public void smokin() throws Exception { + " members"); System.out.println("******"); System.out.println(); - var oracle = domains.get(0).getDelphi(); + var oracle = domains.getFirst().getDelphi(); oracle.add(new Oracle.Namespace("test")).get(); DomainTest.smoke(oracle); } @@ -192,10 +191,7 @@ private Builder params() { .setNumberOfEpochs(12) .setEpochLength(33)) .build()) - .setCheckpointBlockDelta(200) - .setDrainPolicy(ExponentialBackoffPolicy.newBuilder() - .setInitialBackoff(Duration.ofMillis(1)) - .setMaxBackoff(Duration.ofMillis(1))); + .setCheckpointBlockDelta(200); return template; } } diff --git a/sql-state/src/test/java/com/salesforce/apollo/state/AbstractLifecycleTest.java b/sql-state/src/test/java/com/salesforce/apollo/state/AbstractLifecycleTest.java index de2960e19..15948e7ca 100644 --- a/sql-state/src/test/java/com/salesforce/apollo/state/AbstractLifecycleTest.java +++ b/sql-state/src/test/java/com/salesforce/apollo/state/AbstractLifecycleTest.java @@ -380,7 +380,6 @@ private Builder parameters(Context context) { .setCheckpointBlockDelta(checkpointBlockSize()) .setCheckpointSegmentSize(128); - params.getDrainPolicy().setInitialBackoff(Duration.ofMillis(1)).setMaxBackoff(Duration.ofMillis(1)); params.getProducer().ethereal().setNumberOfEpochs(2).setEpochLength(20); return params;