Add scaffolding to start FSC in arbitrary stages without code changes.

Note: there is no config wired in, so it's effectively hardcoded to always start at PLAY or LOGIN
(effectively, immediately), depending on if it's connecting to an old Fireblanket server, or a matching server.
Some versions may still yield a crash & kick due to desynced packet expectations, but this aims to clear that up.

The client will pick the earliest possible entry point presented by the server.

The server currently has no ordering guarantees, and may select any random order at boot time.

Should FSC be broken by a port or misbehaving mod, or partially disabled by SCRAM:
- The client will always fail to understand the packet and proceed as if FSC did not exist.
- The server will always disconnect to any request that was understood.

Author: Ampflower

src/main/java/net/modfest/fireblanket/Fireblanket.java

@@ -24,6 +24,7 @@
 import net.minecraft.core.Registry;
 import net.minecraft.core.registries.BuiltInRegistries;
 import net.minecraft.network.Connection;
+import net.minecraft.network.chat.Component;
 import net.minecraft.network.protocol.Packet;
 import net.minecraft.resources.Identifier;
 import net.minecraft.server.level.ServerLevel;
@@ -51,6 +52,7 @@
 import net.modfest.fireblanket.net.BatchedBEUpdatePayload;
 import net.modfest.fireblanket.net.BatchedEntityVelocityUpdatePacket;
 import net.modfest.fireblanket.net.CommandBlockPacket;
+import net.modfest.fireblanket.net.NetworkState;
 import net.modfest.fireblanket.util.LinkedBlocQueue;
 import net.modfest.fireblanket.world.ItemBan;
 import net.modfest.fireblanket.world.blocks.UpdateSignBlockEntityTypes;
@@ -60,6 +62,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.LockSupport;
 import java.util.function.Consumer;
@@ -190,14 +193,49 @@ public void onInitialize() {
 			ServerLoginConnectionEvents.QUERY_START.addPhaseOrdering(Identifier.parse("fireblanket:pre"), Event.DEFAULT_PHASE);
 			ServerLoginConnectionEvents.QUERY_START.register(Identifier.parse("fireblanket:pre"), (handler, server, sender, synchronizer) -> {
 				if (!server.isSingleplayer()) {
-					sender.sendPacket(FULL_STREAM_COMPRESSION, FriendlyByteBufs.empty());
+					final var buf = FriendlyByteBufs.create();
+					buf.writeCollection(NetworkState.VALID, NetworkState.CODEC);
+					sender.sendPacket(FULL_STREAM_COMPRESSION, buf);
 				}
 			});
 		}
 
 		ServerLoginNetworking.registerGlobalReceiver(FULL_STREAM_COMPRESSION, (server, handler, understood, buf, synchronizer, responseSender) -> {
-			if (understood) {
-				((FSCConnection) ((ServerLoginNetworkHandlerAccessor) handler).fireblanket$getConnection()).fireblanket$enableFullStreamCompression();
+			if (!understood) {
+				return;
+			}
+
+			if (!((((ServerLoginNetworkHandlerAccessor) handler).fireblanket$getConnection()) instanceof FSCConnection connection)) {
+				responseSender.disconnect(Component.translatableWithFallback("fireblanket.fsc.broken", "FSC Broken"));
+				return;
+			}
+
+			if (!buf.isReadable()) {
+				// Old Fireblanket logic; only relevant to backports,
+				// or cases of "why are you using ViaVersion with Fireblanket's networking?".
+				connection.fireblanket$enableFullStreamCompression(NetworkState.PLAY);
+				return;
+			}
+
+			final int len = buf.readByte() & 255;
+
+			if (len > 127 || !buf.isReadable(len)) {
+				responseSender.disconnect(Component.translatableWithFallback("fireblanket.fsc.invalid.size", "FSC Invalid: %s > %s || %1$s > 127", len, buf.readableBytes()));
+			}
+
+			final String rawState = buf.readString(len, StandardCharsets.UTF_8);
+			final NetworkState state = NetworkState.parse(rawState);
+
+			if (!NetworkState.VALID.contains(state)) {
+				// If the client picked a state that is not valid, we cannot continue as we'll crash the client.
+				responseSender.disconnect(Component.translatableWithFallback("fireblanket.fsc.invalid.name", "FSC Invalid: %s", state));
+				return;
+			}
+
+			connection.fireblanket$enableFullStreamCompression(state);
+
+			if (state == NetworkState.LOGIN) {
+				connection.fireblanket$startFullStreamCompression(NetworkState.LOGIN, 0);
 			}
 		});
 

src/main/java/net/modfest/fireblanket/FireblanketClient.java

@@ -29,10 +29,14 @@
 import net.modfest.fireblanket.net.BatchedBEUpdatePayload;
 import net.modfest.fireblanket.net.BatchedEntityVelocityUpdatePacket;
 import net.modfest.fireblanket.net.CommandBlockPacket;
+import net.modfest.fireblanket.net.NetworkState;
 import net.modfest.fireblanket.net.VelocityUpdate;
 import net.modfest.fireblanket.world.render_regions.RegionSyncRequest;
 import net.modfest.fireblanket.world.render_regions.RenderRegions;
 
+import java.nio.charset.StandardCharsets;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 
 public class FireblanketClient implements ClientModInitializer {
@@ -61,12 +65,48 @@ public void onInitializeClient() {
 		});
 
 		ClientLoginNetworking.registerGlobalReceiver(Fireblanket.FULL_STREAM_COMPRESSION, (client, handler, buf, listenerAdder) -> {
-			if (Fireblanket.CAN_USE_ZSTD) {
-				((FSCConnection) ((ClientLoginNetworkHandlerAccessor) handler).fireblanket$getConnection()).fireblanket$enableFullStreamCompression();
+			if (!Fireblanket.CAN_USE_ZSTD) {
+				return CompletableFuture.completedFuture(null);
+			}
+			if (!(((ClientLoginNetworkHandlerAccessor) handler).fireblanket$getConnection() instanceof FSCConnection connection)) {
+				Fireblanket.LOGGER.error("FSC mixins are broken; ZSTD is allowed but FSCConnection is missing?");
+				return CompletableFuture.completedFuture(null);
+			}
+
+			if (!buf.isReadable()) {
+				// We're talking to an older Fireblanket server. Enable at play.
+				connection.fireblanket$enableFullStreamCompression(NetworkState.PLAY);
+				// This also means send an empty buffer back.
 				return CompletableFuture.completedFuture(FriendlyByteBufs.empty());
-			} else {
+			}
+
+			final Set<NetworkState> states = buf.readCollection(HashSet::new, NetworkState.CODEC);
+
+			// Retain all valid states.
+			states.retainAll(NetworkState.VALID);
+
+			if (states.isEmpty()) {
+				// Signal that we did not understand and proceed.
 				return CompletableFuture.completedFuture(null);
 			}
+
+			NetworkState state = NetworkState.UNKNOWN;
+			for (final NetworkState next : states) {
+				if (state.ordinal() > next.ordinal()) {
+					state = next;
+				}
+			}
+
+			connection.fireblanket$enableFullStreamCompression(state);
+
+			if (state == NetworkState.LOGIN) {
+				listenerAdder.accept(_ -> connection.fireblanket$startFullStreamCompression(NetworkState.LOGIN, 0));
+			}
+
+			final var ret = FriendlyByteBufs.create();
+			ret.writeByte(state.getNetworkName().length());
+			ret.writeCharSequence(state.getNetworkName(), StandardCharsets.UTF_8);
+			return CompletableFuture.completedFuture(ret);
 		});
 
 		ClientPlayNetworking.registerGlobalReceiver(BatchedBEUpdatePayload.ID, (payload, context) -> {

src/main/java/net/modfest/fireblanket/mixin/fsc/MixinClientConfigurationNetworkHandler.java

@@ -6,6 +6,7 @@
 import net.minecraft.client.multiplayer.CommonListenerCookie;
 import net.minecraft.network.Connection;
 import net.modfest.fireblanket.mixinsupport.FSCConnection;
+import net.modfest.fireblanket.net.NetworkState;
 import org.spongepowered.asm.mixin.Mixin;
 import org.spongepowered.asm.mixin.injection.At;
 import org.spongepowered.asm.mixin.injection.Inject;
@@ -28,6 +29,6 @@ protected MixinClientConfigurationNetworkHandler(final Minecraft client, final C
 	 **/
 	@Inject(method = "handleConfigurationFinished", at = @At("RETURN"))
 	private void fireblanket$startFSC(CallbackInfo ci) {
-		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(0L);
+		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(NetworkState.PLAY, 0L);
 	}
 }

src/main/java/net/modfest/fireblanket/mixin/fsc/MixinClientConnection.java

@@ -15,13 +15,15 @@
 import net.modfest.fireblanket.Fireblanket;
 import net.modfest.fireblanket.Fireblanket.QueuedPacket;
 import net.modfest.fireblanket.mixinsupport.FSCConnection;
+import net.modfest.fireblanket.net.NetworkState;
 import net.modfest.fireblanket.net.ZstdDecoder;
 import net.modfest.fireblanket.net.ZstdEncoder;
 import net.modfest.fireblanket.util.LinkedBlocQueue;
 import net.modfest.fireblanket.util.ReassignableOutputStream;
 import org.jetbrains.annotations.Nullable;
 import org.spongepowered.asm.mixin.Mixin;
 import org.spongepowered.asm.mixin.Shadow;
+import org.spongepowered.asm.mixin.Unique;
 import org.spongepowered.asm.mixin.injection.At;
 import org.spongepowered.asm.mixin.injection.Inject;
 import org.spongepowered.asm.mixin.injection.Redirect;
@@ -48,8 +50,11 @@ private void sendPacket(Packet<?> packet, ChannelFutureListener callbacks, boole
 	@Shadow
 	public abstract void flushChannel();
 
+	@Unique
 	private final LinkedBlocQueue<QueuedPacket> fireblanket$queue = Fireblanket.getNextQueue();
-	private boolean fireblanket$fsc = false;
+	@Unique
+	private NetworkState fireblanket$fsc = NetworkState.NONE;
+	@Unique
 	private boolean fireblanket$fscStarted = false;
 
 	/**
@@ -85,7 +90,7 @@ private void sendPacket(Packet<?> packet, ChannelFutureListener callbacks, boole
 		// idk man
 		if (packet instanceof ClientboundDisconnectPacket || packet instanceof ClientboundLoginDisconnectPacket) {
 			fireblanket$fscStarted = false;
-			fireblanket$fsc = false;
+			fireblanket$fsc = NetworkState.NONE;
 		}
 	}
 
@@ -100,13 +105,15 @@ private void sendPacket(Packet<?> packet, ChannelFutureListener callbacks, boole
 	 * {@inheritDoc}
 	 */
 	@Override
-	public void fireblanket$startFullStreamCompression(final long millis) {
-		if (!this.fireblanket$fsc) {
+	public void fireblanket$startFullStreamCompression(final NetworkState state, final long millis) {
+		if (state.ordinal() < this.fireblanket$fsc.ordinal()) {
 			return;
 		}
 		if (!this.fireblanket$fscStarted) {
+			Fireblanket.LOGGER.debug("Starting FSC now: {} @ {}ms", state, millis);
 			this.fireblanket$enableFSCNow(millis);
 		} else {
+			Fireblanket.LOGGER.debug("Modifying FSC flush time: {} @ {}ms", state, millis);
 			this.fireblanket$modifyFSC(millis);
 		}
 	}
@@ -151,7 +158,8 @@ private void sendPacket(Packet<?> packet, ChannelFutureListener callbacks, boole
 	}
 
 	@Override
-	public void fireblanket$enableFullStreamCompression() {
-		fireblanket$fsc = true;
+	public void fireblanket$enableFullStreamCompression(final NetworkState state) {
+		Fireblanket.LOGGER.debug("Enabling FSC: {}", state);
+		this.fireblanket$fsc = state;
 	}
 }

src/main/java/net/modfest/fireblanket/mixin/fsc/MixinClientLoginPacketListenerImpl.java

@@ -3,6 +3,7 @@
 import net.minecraft.client.multiplayer.ClientHandshakePacketListenerImpl;
 import net.minecraft.network.Connection;
 import net.modfest.fireblanket.mixinsupport.FSCConnection;
+import net.modfest.fireblanket.net.NetworkState;
 import org.spongepowered.asm.mixin.Final;
 import org.spongepowered.asm.mixin.Mixin;
 import org.spongepowered.asm.mixin.Shadow;
@@ -33,6 +34,6 @@ public class MixinClientLoginPacketListenerImpl {
 		)
 	)
 	private void fireblanket$startFSC(CallbackInfo ci) {
-		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(0L);
+		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(NetworkState.CONFIGURATION, 0L);
 	}
 }

src/main/java/net/modfest/fireblanket/mixin/fsc/MixinServerConfigurationNetworkHandler.java

@@ -6,6 +6,7 @@
 import net.minecraft.server.network.ServerCommonPacketListenerImpl;
 import net.minecraft.server.network.ServerConfigurationPacketListenerImpl;
 import net.modfest.fireblanket.mixinsupport.FSCConnection;
+import net.modfest.fireblanket.net.NetworkState;
 import org.spongepowered.asm.mixin.Mixin;
 import org.spongepowered.asm.mixin.injection.At;
 import org.spongepowered.asm.mixin.injection.Inject;
@@ -28,6 +29,6 @@ public MixinServerConfigurationNetworkHandler(final MinecraftServer server, fina
 	 **/
 	@Inject(method = "handleConfigurationFinished", at = @At(value = "INVOKE", target = "Lnet/minecraft/network/Connection;setupOutboundProtocol(Lnet/minecraft/network/ProtocolInfo;)V"))
 	private void fireblanket$startFSC(CallbackInfo ci) {
-		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(40L);
+		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(NetworkState.PLAY, 40L);
 	}
 }

src/main/java/net/modfest/fireblanket/mixin/fsc/MixinServerLoginPacketListenerImpl.java

@@ -3,6 +3,7 @@
 import net.minecraft.network.Connection;
 import net.minecraft.server.network.ServerLoginPacketListenerImpl;
 import net.modfest.fireblanket.mixinsupport.FSCConnection;
+import net.modfest.fireblanket.net.NetworkState;
 import org.spongepowered.asm.mixin.Final;
 import org.spongepowered.asm.mixin.Mixin;
 import org.spongepowered.asm.mixin.Shadow;
@@ -27,6 +28,6 @@ public class MixinServerLoginPacketListenerImpl {
 	 **/
 	@Inject(method = "handleLoginAcknowledgement", at = @At(value = "INVOKE", target = "Lnet/minecraft/network/Connection;setupOutboundProtocol(Lnet/minecraft/network/ProtocolInfo;)V"))
 	private void fireblanket$startFSC(CallbackInfo ci) {
-		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(0L);
+		((FSCConnection) this.connection).fireblanket$startFullStreamCompression(NetworkState.CONFIGURATION, 0L);
 	}
 }

src/main/java/net/modfest/fireblanket/mixinsupport/FSCConnection.java

@@ -8,6 +8,7 @@
 import net.modfest.fireblanket.mixin.fsc.MixinClientLoginPacketListenerImpl;
 import net.modfest.fireblanket.mixin.fsc.MixinServerConfigurationNetworkHandler;
 import net.modfest.fireblanket.mixin.fsc.MixinServerLoginPacketListenerImpl;
+import net.modfest.fireblanket.net.NetworkState;
 
 public interface FSCConnection {
 
@@ -31,8 +32,13 @@ public interface FSCConnection {
 	 * @see MixinClientLoginPacketListenerImpl
 	 * @see MixinServerLoginPacketListenerImpl
 	 */
-	void fireblanket$startFullStreamCompression(final long millis);
+	void fireblanket$startFullStreamCompression(final NetworkState state, final long millis);
 
-	void fireblanket$enableFullStreamCompression();
+	/**
+	 * Sets the starting point for full-stream compression.
+	 *
+	 * @implSpec The client and server <em>must</em> agree, else you will have decoding errors.
+	 */
+	void fireblanket$enableFullStreamCompression(final NetworkState state);
 
 }

src/main/java/net/modfest/fireblanket/net/NetworkState.java

@@ -0,0 +1,64 @@
+package net.modfest.fireblanket.net;
+
+import io.netty.buffer.ByteBuf;
+import net.minecraft.network.codec.ByteBufCodecs;
+import net.minecraft.network.codec.StreamCodec;
+
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * Allows negotiating the network state required to enable FSC.
+ *
+ * @author Ampflower
+ * @implNote The order of the enum is load bearing, and expected to be sequential.
+ * 	If this ever requires changes, please be sure to order all new enums to be <em>before</em> {@link #NONE},
+ * 	and that the enums are ordered to how they'll be encountered in the network protocol.
+ * 	<p>
+ * 	If the entrypoint of a given enum has been moved in such a way that starting FSC is now
+ * 	incompatible with the previous version, give it a new network name to prevent the old client from using it.
+ **/
+public enum NetworkState {
+	// This would ideally be via handshake, using a special subdomain that is sent by the client,
+	// if it sees a magic in the status. This would only be possible if the server has been seen before joining.
+	// This can break proxies, and may be disableable for proxies.
+	// However, if it is seen at login stage instead, this can be effectively be considered as enable immediately.
+	// Since login runs in lockstep, we can safely swap the pipeline pieces to enable FSC.
+	LOGIN("login"),
+	CONFIGURATION("configuration"),
+	PLAY("play"),
+	// Only be possible on a vanilla client
+	// A Fireblanket client sending this should be disconnected?
+	// The client must refuse to acknowledge the packet if it receives this state explicitly.
+	NONE("none"),
+	// acts as PLAY if the server can't understand it.
+	// The client must refuse to acknowledge the packet if it receives this state explicitly.
+	UNKNOWN("unknown"),
+	;
+
+	public static final StreamCodec<ByteBuf, NetworkState> CODEC = ByteBufCodecs.STRING_UTF8
+		.map(NetworkState::parse, NetworkState::getNetworkName);
+
+	// TODO: configuration-based negotiation
+	public static final Set<NetworkState> VALID = Set.of(LOGIN, CONFIGURATION, PLAY);
+
+	private final String name;
+
+	NetworkState(final String name) {
+		this.name = name;
+	}
+
+	public static NetworkState parse(final String string) {
+		return switch (string.toLowerCase(Locale.ROOT)) {
+			case "none" -> NONE;
+			case "login" -> LOGIN;
+			case "configuration" -> CONFIGURATION;
+			case "play" -> PLAY;
+			default -> UNKNOWN;
+		};
+	}
+
+	public final String getNetworkName() {
+		return name;
+	}
+}

src/main/resources/assets/fireblanket/lang/en_us.json

@@ -7,6 +7,9 @@
 	"gamerule.fireblanket:lightningBroadcastRadius.description": "How far in blocks players will receive lightning strikes from in-game. Set to 0 to effectively disable broadcasting.",
 	"commands.gamerule.locked": "You do not have permission to modify this gamerule",
 	"commandsBlock.commandSetByPlayer": "%s set the command at (%s) to: %s",
+	"fireblanket.fsc.broken": "[Fireblanket] Despite relevant mixins being disabled or broken, we're still requesting FSC?",
+	"fireblanket.fsc.invalid.size": "[Fireblanket] Illegal or invalid string size: %s; exceeded 127 or readable bytes of: %s",
+	"fireblanket.fsc.invalid.name": "[Fireblanket] Your client gave an invalid FSC entrypoint: %s",
 	"fireblanket.commands.command.grep.entry": "[%s] [Creator: %s] [Updater: %s] [%s]\n> %s",
 	"fireblanket.commands.command.grep.entry.same": "[%s] [%s] [%s]\n> %s",
 	"fireblanket.commands.command.grep.result": "Found %s matches.",