chore(1416): Publisher cleanup items #1490

jsync-swirlds · 2025-08-13T01:12:44Z

Reviewer Notes

Implemented proper condition wait for the publisher manager when there are no incoming data items to forward to messaging.
Added metrics for closed blocks (complete and incomplete).
Improved queue transfers
Added a node level configuration, and used that for "earliest managed block".

Related Issue(s)

Partial resolution to #1416.

ata-nas

Great improvements @jsync-swirlds! Looking forward to seeing more 🙌

...ublisher/src/main/java/org/hiero/block/node/stream/publisher/LiveStreamPublisherManager.java

...e/stream-publisher/src/main/java/org/hiero/block/node/stream/publisher/PublisherHandler.java

AlfredoG87

LGTM 👍

Nana-EC

Looks good

Nana-EC · 2025-08-15T03:27:47Z

...e/stream-publisher/src/main/java/org/hiero/block/node/stream/publisher/PublisherHandler.java

+    /**
+     * todo(1420) add documentation
+     */
+    private void sendEndAndReset(final Code endOfStreamCode) {


nit: without docs this reads as send End and Reset codes.
Rather we're sending an end and resetting State.

Suggested change

private void sendEndAndReset(final Code endOfStreamCode) {

private void sendEndAndResetState(final Code endOfStreamCode) {

or

Suggested change

private void sendEndAndReset(final Code endOfStreamCode) {

private void sendEndThenResetState(final Code endOfStreamCode) {

I'd suggest:

Suggested change

private void sendEndAndReset(final Code endOfStreamCode) {

private void sendEndAndShutdown(final Code endOfStreamCode) {

As we do seem to call shutdown there.

It's impossible to send two codes at once.
That said, sendEndAndResetState makes sense.

It's impossible to send two codes at once. That said, sendEndAndResetState makes sense.

I do not seem to get what you mean here? We do call shutdown in the method. Shutdown does not send any responses.

The original question was that the name seems to claim it's sending two codes, which I can see, but it's also a non-sequitur, because nothing can send two response codes simultaneously.

I've changed the name to sendEndAndResetState

ata-nas

We need the changes from #1497 and we need to fix non-compiling tests. Will rebase after #1497 merges.

ata-nas · 2025-08-15T16:28:03Z

...e/stream-publisher/src/main/java/org/hiero/block/node/stream/publisher/PublisherHandler.java

+     * to force all handlers to close their publisher communication channels.
+     */
+    public void closeCommunication() {
+        sendEndOfStream(Code.SUCCESS);


we should shutdown() here as well?

The important thing is to probably call onComplete(), the manager will clear the rest (remove handler) since it's shutting down.

Maybe we aim to call the other method instead? It will shutdown, which will call onComplete as well (since #1497).

Suggested change

sendEndOfStream(Code.SUCCESS);

sendEndAndReset(Code.SUCCESS);

I very deliberately did not shutdown or call onComplete here, as it creates a race condition with the manager accomplishing much the same tasks.

Ah, I see. We should shutdown the connections at some point however.

P.S. please beware that with the introduction of #1497, the shutdown() method of the handler will call onComplete().

Yes, shutdown should call replies.onShutdown, just need to be careful with idempotence.

ata-nas

Cleanup suggestions

ata-nas · 2025-08-18T08:23:40Z

...ublisher/src/main/java/org/hiero/block/node/stream/publisher/LiveStreamPublisherManager.java

+        // Cancel the queue forwarder task if it is running.
+        if (queueForwarderResult != null) {
+            queueForwarderResult.cancel(true);
+            queueForwarderResult = null;
+        }


Shouldn't we cancel the forwarder task first and then proceed to do all shutdowns?

shutdowns should happen before the forwarder task ends, so we don't lose already-received-and-completed blocks.

I see, thanks for clarifying. Was wondering, won't that interfere with future acknowledgements however? Cause it is the manager that listens for the notifications. If we shut it down, acks will not happen.

ata-nas · 2025-08-18T08:24:40Z

...ublisher/src/main/java/org/hiero/block/node/stream/publisher/LiveStreamPublisherManager.java

+        for (final Long nextKey : handlers.keySet()) {
+            final PublisherHandler value = handlers.remove(nextKey);
+            if (value != null) {
+                value.closeCommunication();
+            }
+        }


Would it be beneficial to fork all of these close communications, after all, we take time to attempt to send responses? (Parallel stream)

Possibly, but we have to wait for the full set to complete.
As this is part of shutdown, it isn't strictly necessary to be high performance and correctness is perhaps more imporant.
We may want to revisit this in follow-up.

ata-nas · 2025-08-18T08:26:24Z

...ublisher/src/main/java/org/hiero/block/node/stream/publisher/LiveStreamPublisherManager.java

@@ -472,11 +549,12 @@ private void updateBlockNumbers(final BlockNodeContext serverContext) {
        // Always set the last persisted block number, even if there are no
        // known blocks.
        lastPersistedBlockNumber.set(latestKnownBlock);
+        NodeConfig nodeConfiguration = serverContext.configuration().getConfigData(NodeConfig.class);


The NodeConfig seems to not be properly exported. Unit tests fail and starting up the node also fails because the config record is not found.

E.G. when running a test:

No config data record available of type 'class org.hiero.block.node.app.config.node.NodeConfig' java.lang.IllegalArgumentException: No config data record available of type 'class org.hiero.block.node.app.config.node.NodeConfig' at com.swirlds.config.impl@0.61.3/com.swirlds.config.impl.internal.ConfigDataService.getConfigData(ConfigDataService.java:96) at com.swirlds.config.impl@0.61.3/com.swirlds.config.impl.internal.ConfigurationImpl.getConfigData(ConfigurationImpl.java:211) at org.hiero.block.node.stream.publisher@0.17.0-SNAPSHOT/org.hiero.block.node.stream.publisher.LiveStreamPublisherManager.updateBlockNumbers(LiveStreamPublisherManager.java:552) at org.hiero.block.node.stream.publisher@0.17.0-SNAPSHOT/org.hiero.block.node.stream.publisher.LiveStreamPublisherManager.<init>(LiveStreamPublisherManager.java:94) at org.hiero.block.node.stream.publisher@0.17.0-SNAPSHOT/org.hiero.block.node.stream.publisher.LiveStreamPublisherManagerTest$FunctionalityTests.setup(LiveStreamPublisherManagerTest.java:165) at java.base/java.lang.reflect.Method.invoke(Method.java:580) at java.base/java.util.Optional.ifPresent(Optional.java:178) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184) at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184) at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) at java.base/java.util.Iterator.forEachRemaining(Iterator.java:133) at java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1939) at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174) at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596) at java.base/java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:276) at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1708) at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174) at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596) at java.base/java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:276) at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1708) at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174) at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596) at java.base/java.util.ArrayList.forEach(ArrayList.java:1596) at java.base/java.util.ArrayList.forEach(ArrayList.java:1596) at java.base/java.util.ArrayList.forEach(ArrayList.java:1596) at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)

ata-nas · 2025-08-18T08:27:32Z

...e/stream-publisher/src/main/java/org/hiero/block/node/stream/publisher/PublisherHandler.java

+    /**
+     * todo(1420) add documentation
+     */
+    private void sendEndAndReset(final Code endOfStreamCode) {


I'd suggest:

Suggested change

private void sendEndAndReset(final Code endOfStreamCode) {

private void sendEndAndShutdown(final Code endOfStreamCode) {

As we do seem to call shutdown there.

ata-nas · 2025-08-18T08:30:03Z

@jsync-swirlds FYI I've rebased this branch on top of main, where we had a big merge conflict.

cc: @Nana-EC @AlfredoG87

* Implemented proper condition wait for the publisher manager when there are no incoming data items to forward to messaging. * Added metrics for closed blocks (complete and incomplete). * Improved queue transfers * Added a node level configuration, and used that for "earliest managed block". Signed-off-by: Joseph S <121976561+jsync-swirlds@users.noreply.github.com> Signed-off-by: Atanas Atanasov <a.v.atanasov98@gmail.com>

jsync-swirlds added this to the 0.17.0 milestone Aug 13, 2025

jsync-swirlds requested review from ata-nas and AlfredoG87 August 13, 2025 01:12

jsync-swirlds self-assigned this Aug 13, 2025

jsync-swirlds added the Publisher Plugin Issue related to Publisher Plugin label Aug 13, 2025

jsync-swirlds force-pushed the 1416-publisher-cleanup branch from 3898ed5 to 43b6e5b Compare August 13, 2025 01:39

ata-nas reviewed Aug 13, 2025

View reviewed changes

jsync-swirlds force-pushed the 1416-publisher-cleanup branch 2 times, most recently from 9b30363 to 1183720 Compare August 13, 2025 23:52

jsync-swirlds marked this pull request as ready for review August 13, 2025 23:53

jsync-swirlds requested review from a team as code owners August 13, 2025 23:53

jsync-swirlds requested review from Nana-EC and ata-nas August 13, 2025 23:53

jsync-swirlds force-pushed the 1416-publisher-cleanup branch from 1183720 to c6385a9 Compare August 13, 2025 23:56

AlfredoG87 previously approved these changes Aug 14, 2025

View reviewed changes

Nana-EC previously approved these changes Aug 15, 2025

View reviewed changes

ata-nas requested changes Aug 15, 2025

View reviewed changes

ata-nas reviewed Aug 15, 2025

View reviewed changes

ata-nas dismissed stale reviews from Nana-EC and AlfredoG87 via 18cc999 August 18, 2025 08:09

ata-nas force-pushed the 1416-publisher-cleanup branch 2 times, most recently from 18cc999 to e4420e7 Compare August 18, 2025 08:19

ata-nas reviewed Aug 18, 2025

View reviewed changes

jsync-swirlds force-pushed the 1416-publisher-cleanup branch from e4420e7 to 7a083ab Compare August 18, 2025 22:19

	private void sendEndAndReset(final Code endOfStreamCode) {
	private void sendEndAndResetState(final Code endOfStreamCode) {

	private void sendEndAndReset(final Code endOfStreamCode) {
	private void sendEndThenResetState(final Code endOfStreamCode) {

	private void sendEndAndReset(final Code endOfStreamCode) {
	private void sendEndAndShutdown(final Code endOfStreamCode) {

	sendEndOfStream(Code.SUCCESS);
	sendEndAndReset(Code.SUCCESS);

chore(1416): Publisher cleanup items #1490

Are you sure you want to change the base?

chore(1416): Publisher cleanup items #1490

Conversation

jsync-swirlds commented Aug 13, 2025

Reviewer Notes

Related Issue(s)

Uh oh!

ata-nas left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

AlfredoG87 left a comment

Choose a reason for hiding this comment

Uh oh!

Nana-EC left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ata-nas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ata-nas Aug 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ata-nas left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ata-nas Aug 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jsync-swirlds Aug 18, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ata-nas commented Aug 18, 2025

Uh oh!

Uh oh!

ata-nas Aug 18, 2025 •

edited

Loading

ata-nas Aug 18, 2025 •

edited

Loading

jsync-swirlds Aug 18, 2025 •

edited

Loading