Add large attachment syncing

fabric-data-attachment-api-v1/src/client/java/net/fabricmc/fabric/impl/attachment/client/AttachmentSyncClient.java

@@ -20,7 +20,6 @@
 import net.fabricmc.fabric.api.client.networking.v1.ClientConfigurationNetworking;
 import net.fabricmc.fabric.api.client.networking.v1.ClientPlayNetworking;
 import net.fabricmc.fabric.impl.attachment.AttachmentEntrypoint;
-import net.fabricmc.fabric.impl.attachment.sync.AttachmentChange;
 import net.fabricmc.fabric.impl.attachment.sync.AttachmentSync;
 import net.fabricmc.fabric.impl.attachment.sync.AttachmentSyncException;
 import net.fabricmc.fabric.impl.attachment.sync.clientbound.ClientboundAttachmentSyncPayload;
@@ -37,16 +36,13 @@ public void onInitializeClient() {
 
 		// play
 		ClientPlayNetworking.registerGlobalReceiver(
-				ClientboundAttachmentSyncPayload.ID,
+				ClientboundAttachmentSyncPayload.TYPE,
 				(payload, context) -> {
-					for (AttachmentChange attachmentChange : payload.attachments()) {
-						try {
-							attachmentChange.tryApply(context.client().level);
-						} catch (AttachmentSyncException e) {
-							AttachmentEntrypoint.LOGGER.error("Error accepting attachment changes", e);
-							context.responseSender().disconnect(e.getComponent());
-							break;
-						}
+					try {
+						payload.attachment().tryApply(context.client().level);
+					} catch (AttachmentSyncException e) {
+						AttachmentEntrypoint.LOGGER.error("Error accepting attachment changes", e);
+						context.responseSender().disconnect(e.getComponent());
 					}
 				}
 		);

fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/api/attachment/v1/AttachmentRegistry.java

@@ -159,6 +159,18 @@ public interface Builder<A> {
 		 */
 		AttachmentRegistry.Builder<A> syncWith(StreamCodec<? super RegistryFriendlyByteBuf, A> streamCodec, AttachmentSyncPredicate syncPredicate);
 
+		/**
+		 * Declares that this attachment type may be automatically synchronized with some clients, as determined by {@code syncPredicate}.
+		 *
+		 * <p>The max size limit should be increased with care, as syncing large amounts of data may result in network lag and excessive bandwidth usage.
+		 *
+		 * @param streamCodec the codec used to serialize the attachment data over the network
+		 * @param syncPredicate an {@link AttachmentSyncPredicate} determining with which clients to synchronize data
+		 * @param maxSyncSize the max number of data bytes that can be synced, defaults to 1 MiB minus some small padding
+		 * @return the builder
+		 */
+		AttachmentRegistry.Builder<A> syncWith(StreamCodec<? super RegistryFriendlyByteBuf, A> streamCodec, AttachmentSyncPredicate syncPredicate, int maxSyncSize);
+
 		/**
 		 * Builds and registers the {@link AttachmentType}.
 		 *

fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/impl/attachment/AttachmentRegistryImpl.java

@@ -43,6 +43,7 @@ public final class AttachmentRegistryImpl {
 	private static final Map<Identifier, AttachmentType<?>> attachmentRegistry = new HashMap<>();
 	private static final Set<Identifier> syncableAttachments = new HashSet<>();
 	private static final Set<Identifier> syncableView = Collections.unmodifiableSet(syncableAttachments);
+	private static int maxSyncPacketSize = AttachmentSync.DEFAULT_ATTACHMENT_SYNC_PACKET_SIZE;
 
 	public static <A> void register(Identifier id, AttachmentType<A> attachmentType) {
 		AttachmentType<?> existing = attachmentRegistry.put(id, attachmentType);
@@ -74,6 +75,16 @@ public static <A> AttachmentRegistry.Builder<A> builder() {
 		return new BuilderImpl<>();
 	}
 
+	public static int getMaxSyncPacketSize() {
+		if (maxSyncPacketSize == -1) {
+			throw new IllegalStateException("getMaxSyncPacketSize should only be called ONCE!");
+		}
+
+		int maxSize = maxSyncPacketSize;
+		maxSyncPacketSize = -1;
+		return maxSize;
+	}
+
 	public static class BuilderImpl<A> implements AttachmentRegistry.Builder<A> {
 		@Nullable
 		private Supplier<A> defaultInitializer = null;
@@ -84,6 +95,7 @@ public static class BuilderImpl<A> implements AttachmentRegistry.Builder<A> {
 		@Nullable
 		private AttachmentSyncPredicate syncPredicate = null;
 		private boolean copyOnDeath = false;
+		private int maxSyncSize = -1;
 
 		@Override
 		public AttachmentRegistry.Builder<A> persistent(Codec<A> codec) {
@@ -107,7 +119,7 @@ public AttachmentRegistry.Builder<A> initializer(Supplier<A> initializer) {
 			return this;
 		}
 
-		@Deprecated
+		@Override
 		public AttachmentRegistry.Builder<A> syncWith(StreamCodec<? super RegistryFriendlyByteBuf, A> streamCodec, AttachmentSyncPredicate syncPredicate) {
 			Objects.requireNonNull(streamCodec, "stream codec cannot be null");
 			Objects.requireNonNull(syncPredicate, "sync predicate cannot be null");
@@ -117,6 +129,18 @@ public AttachmentRegistry.Builder<A> syncWith(StreamCodec<? super RegistryFriend
 			return this;
 		}
 
+		@Override
+		public AttachmentRegistry.Builder<A> syncWith(StreamCodec<? super RegistryFriendlyByteBuf, A> streamCodec, AttachmentSyncPredicate syncPredicate, int maxSyncSize) {
+			if (maxSyncSize < 0) {
+				throw new IllegalArgumentException("maxSyncSize must be positive!");
+			}
+
+			syncWith(streamCodec, syncPredicate);
+			this.maxSyncSize = maxSyncSize;
+
+			return this;
+		}
+
 		@Override
 		public AttachmentType<A> buildAndRegister(Identifier id) {
 			Objects.requireNonNull(id, "identifier cannot be null");
@@ -130,13 +154,24 @@ public AttachmentType<A> buildAndRegister(Identifier id) {
 				);
 			}
 
+			if (maxSyncSize <= AttachmentSync.DEFAULT_MAX_DATA_SIZE) {
+				maxSyncSize = AttachmentSync.DEFAULT_MAX_DATA_SIZE;
+			} else if (maxSyncPacketSize == -1) {
+				throw new IllegalStateException("Large attachment " + id + " registered too late! Must be registered during mod initialization.");
+			} else {
+				int newMaxPacketSize = maxSyncSize + AttachmentSync.MAX_PADDING_SIZE_IN_BYTES;
+				newMaxPacketSize = newMaxPacketSize < 0 ? Integer.MAX_VALUE : newMaxPacketSize; // prevent overflow
+				maxSyncPacketSize = Math.max(newMaxPacketSize, maxSyncPacketSize);
+			}
+
 			var attachment = new AttachmentTypeImpl<>(
 					id,
 					defaultInitializer,
 					persistenceCodec,
 					streamCodec,
 					syncPredicate,
-					copyOnDeath
+					copyOnDeath,
+					maxSyncSize
 			);
 			register(id, attachment);
 			return attachment;

fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/impl/attachment/AttachmentTypeImpl.java

@@ -34,7 +34,8 @@ public record AttachmentTypeImpl<A>(
 		@Nullable Codec<A> persistenceCodec,
 		@Nullable StreamCodec<? super RegistryFriendlyByteBuf, A> streamCodec,
 		@Nullable AttachmentSyncPredicate syncPredicate,
-		boolean copyOnDeath
+		boolean copyOnDeath,
+		int maxSyncSize
 ) implements AttachmentType<A> {
 	@Override
 	public boolean isSynced() {

fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/impl/attachment/sync/AttachmentChange.java

@@ -16,11 +16,7 @@
 
 package net.fabricmc.fabric.impl.attachment.sync;
 
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.List;
 import java.util.Objects;
-import java.util.Set;
 
 import io.netty.buffer.Unpooled;
 import org.jspecify.annotations.Nullable;
@@ -35,19 +31,13 @@
 import net.minecraft.network.codec.ByteBufCodecs;
 import net.minecraft.network.codec.StreamCodec;
 import net.minecraft.resources.Identifier;
-import net.minecraft.server.level.ServerPlayer;
 import net.minecraft.world.level.Level;
 
 import net.fabricmc.fabric.api.attachment.v1.AttachmentTarget;
 import net.fabricmc.fabric.api.attachment.v1.AttachmentType;
 import net.fabricmc.fabric.api.networking.v1.FriendlyByteBufs;
-import net.fabricmc.fabric.api.networking.v1.ServerPlayNetworking;
 import net.fabricmc.fabric.impl.attachment.AttachmentRegistryImpl;
 import net.fabricmc.fabric.impl.attachment.AttachmentTypeImpl;
-import net.fabricmc.fabric.impl.attachment.sync.clientbound.ClientboundAttachmentSyncPayload;
-import net.fabricmc.fabric.mixin.attachment.ServerboundCustomPayloadPacketAccessor;
-import net.fabricmc.fabric.mixin.attachment.VarIntAccessor;
-import net.fabricmc.fabric.mixin.networking.accessor.ServerCommonPacketListenerImplAccessor;
 
 public record AttachmentChange(AttachmentTargetInfo<?> targetInfo, AttachmentType<?> type, byte[] data) {
 	public static final StreamCodec<FriendlyByteBuf, AttachmentChange> PACKET_CODEC = StreamCodec.composite(
@@ -59,8 +49,6 @@ public record AttachmentChange(AttachmentTargetInfo<?> targetInfo, AttachmentTyp
 			ByteBufCodecs.BYTE_ARRAY, AttachmentChange::data,
 			AttachmentChange::new
 	);
-	private static final int MAX_PADDING_SIZE_IN_BYTES = AttachmentTargetInfo.MAX_SIZE_IN_BYTES + AttachmentSync.MAX_IDENTIFIER_SIZE;
-	private static final int MAX_DATA_SIZE_IN_BYTES = ServerboundCustomPayloadPacketAccessor.getMaxPayloadSize() - MAX_PADDING_SIZE_IN_BYTES;
 
 	@SuppressWarnings("unchecked")
 	public static AttachmentChange create(AttachmentTargetInfo<?> targetInfo, AttachmentType<?> type, @Nullable Object value, RegistryAccess registryAccess) {
@@ -77,50 +65,22 @@ public static AttachmentChange create(AttachmentTargetInfo<?> targetInfo, Attach
 			buf.writeBoolean(false);
 		}
 
-		byte[] encoded = buf.array();
+		// buf.array() returns the backing array directly, which often contains unused space
+		byte[] encoded = new byte[buf.readableBytes()];
+		buf.readBytes(encoded);
+		int maxDataSize = ((AttachmentTypeImpl<?>) type).maxSyncSize();
 
-		if (encoded.length > MAX_DATA_SIZE_IN_BYTES) {
+		if (encoded.length > maxDataSize) {
 			throw new IllegalArgumentException("Data for attachment '%s' was too big (%d bytes, over maximum %d)".formatted(
 					type.identifier(),
 					encoded.length,
-					MAX_DATA_SIZE_IN_BYTES
+					maxDataSize
 			));
 		}
 
 		return new AttachmentChange(targetInfo, type, encoded);
 	}
 
-	public static void partitionAndSendPackets(List<AttachmentChange> changes, ServerPlayer player) {
-		Set<Identifier> supported = ((SupportedAttachmentsConnection) ((ServerCommonPacketListenerImplAccessor) player.connection).getConnection())
-				.fabric_getSupportedAttachments();
-		// sort by size to better partition packets
-		changes.sort(Comparator.comparingInt(c -> c.data().length));
-		List<AttachmentChange> packetChanges = new ArrayList<>();
-		int maxVarIntSize = VarIntAccessor.getMaxByteSize();
-		int byteSize = maxVarIntSize;
-
-		for (AttachmentChange change : changes) {
-			if (!supported.contains(change.type.identifier())) {
-				continue;
-			}
-
-			int size = MAX_PADDING_SIZE_IN_BYTES + change.data.length;
-
-			if (!packetChanges.isEmpty() && byteSize + size > MAX_DATA_SIZE_IN_BYTES) {
-				ServerPlayNetworking.send(player, new ClientboundAttachmentSyncPayload(List.copyOf(packetChanges)));
-				packetChanges.clear();
-				byteSize = maxVarIntSize;
-			}
-
-			packetChanges.add(change);
-			byteSize += size;
-		}
-
-		if (!packetChanges.isEmpty()) {
-			ServerPlayNetworking.send(player, new ClientboundAttachmentSyncPayload(packetChanges));
-		}
-	}
-
 	@SuppressWarnings("unchecked")
 	@Nullable
 	public Object decodeValue(RegistryAccess registryAccess) {

fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/impl/attachment/sync/AttachmentSync.java

@@ -22,8 +22,13 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
+import io.netty.buffer.ByteBufUtil;
+
 import net.minecraft.network.Connection;
+import net.minecraft.network.VarInt;
 import net.minecraft.network.protocol.Packet;
+import net.minecraft.network.protocol.game.ClientGamePacketListener;
+import net.minecraft.network.protocol.game.ClientboundBundlePacket;
 import net.minecraft.resources.Identifier;
 import net.minecraft.server.level.ServerPlayer;
 import net.minecraft.server.network.ConfigurationTask;
@@ -42,10 +47,22 @@
 import net.fabricmc.fabric.impl.attachment.sync.clientbound.ClientboundAttachmentSyncPayload;
 import net.fabricmc.fabric.impl.attachment.sync.clientbound.ClientboundRequestAcceptedAttachmentsPayload;
 import net.fabricmc.fabric.impl.attachment.sync.serverbound.ServerboundAcceptedAttachmentsPayload;
+import net.fabricmc.fabric.mixin.attachment.ClientboundCustomPayloadPacketAccessor;
 import net.fabricmc.fabric.mixin.networking.accessor.ServerCommonPacketListenerImplAccessor;
 
 public class AttachmentSync implements ModInitializer {
 	public static final int MAX_IDENTIFIER_SIZE = 256;
+	public static final int MAX_PADDING_SIZE_IN_BYTES = AttachmentTargetInfo.MAX_SIZE_IN_BYTES + MAX_IDENTIFIER_SIZE;
+	public static final int DEFAULT_MAX_DATA_SIZE;
+	public static final int DEFAULT_ATTACHMENT_SYNC_PACKET_SIZE;
+
+	static {
+		// ensure no splitting by default
+		int identifierSize = ByteBufUtil.utf8MaxBytes(ClientboundAttachmentSyncPayload.PACKET_ID.toString());
+		int networkingApiPaddingSize = VarInt.getByteSize(identifierSize) + identifierSize + 5 * 2;
+		DEFAULT_MAX_DATA_SIZE = ClientboundCustomPayloadPacketAccessor.getMaxPayloadSize() - MAX_PADDING_SIZE_IN_BYTES - networkingApiPaddingSize;
+		DEFAULT_ATTACHMENT_SYNC_PACKET_SIZE = MAX_PADDING_SIZE_IN_BYTES + DEFAULT_MAX_DATA_SIZE;
+	}
 
 	public static ServerboundAcceptedAttachmentsPayload createResponsePayload() {
 		return new ServerboundAcceptedAttachmentsPayload(AttachmentRegistryImpl.getSyncableAttachments());
@@ -56,7 +73,23 @@ public static void trySync(AttachmentChange change, ServerPlayer player) {
 				.fabric_getSupportedAttachments();
 
 		if (supported.contains(change.type().identifier())) {
-			ServerPlayNetworking.send(player, new ClientboundAttachmentSyncPayload(List.of(change)));
+			ServerPlayNetworking.send(player, new ClientboundAttachmentSyncPayload(change));
+		}
+	}
+
+	public static void trySync(List<AttachmentChange> changes, ServerPlayer player) {
+		Set<Identifier> supported = ((SupportedAttachmentsConnection) ((ServerCommonPacketListenerImplAccessor) player.connection).getConnection())
+				.fabric_getSupportedAttachments();
+
+		List<Packet<? super ClientGamePacketListener>> syncableChanges = new ArrayList<>();
+		changes.forEach(change -> {
+			if (supported.contains(change.type().identifier())) {
+				syncableChanges.add(ServerPlayNetworking.createClientboundPacket(new ClientboundAttachmentSyncPayload(change)));
+			}
+		});
+
+		if (!syncableChanges.isEmpty()) {
+			ServerPlayNetworking.getSender(player).sendPacket(new ClientboundBundlePacket(syncableChanges));
 		}
 	}
 
@@ -105,8 +138,8 @@ public void onInitialize() {
 				});
 
 		// Play
-		PayloadTypeRegistry.clientboundPlay().register(
-				ClientboundAttachmentSyncPayload.ID, ClientboundAttachmentSyncPayload.CODEC);
+		PayloadTypeRegistry.clientboundPlay().registerLarge(
+				ClientboundAttachmentSyncPayload.TYPE, ClientboundAttachmentSyncPayload.CODEC, AttachmentRegistryImpl::getMaxSyncPacketSize);
 
 		ServerPlayerEvents.JOIN.register((player) -> {
 			List<AttachmentChange> changes = new ArrayList<>();
@@ -116,7 +149,7 @@ public void onInitialize() {
 			((AttachmentTargetImpl) player).fabric_computeInitialSyncChanges(player, changes::add);
 
 			if (!changes.isEmpty()) {
-				AttachmentChange.partitionAndSendPackets(changes, player);
+				trySync(changes, player);
 			}
 		});
 
@@ -127,7 +160,7 @@ public void onInitialize() {
 			((AttachmentTargetImpl) destination).fabric_computeInitialSyncChanges(player, changes::add);
 
 			if (!changes.isEmpty()) {
-				AttachmentChange.partitionAndSendPackets(changes, player);
+				trySync(changes, player);
 			}
 		});
 
@@ -136,7 +169,7 @@ public void onInitialize() {
 			((AttachmentTargetImpl) trackedEntity).fabric_computeInitialSyncChanges(player, changes::add);
 
 			if (!changes.isEmpty()) {
-				AttachmentChange.partitionAndSendPackets(changes, player);
+				trySync(changes, player);
 			}
 		});
 	}

fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/impl/attachment/sync/clientbound/ClientboundAttachmentSyncPayload.java

@@ -16,26 +16,24 @@
 
 package net.fabricmc.fabric.impl.attachment.sync.clientbound;
 
-import java.util.List;
-
 import net.minecraft.network.FriendlyByteBuf;
-import net.minecraft.network.codec.ByteBufCodecs;
 import net.minecraft.network.codec.StreamCodec;
 import net.minecraft.network.protocol.common.custom.CustomPacketPayload;
 import net.minecraft.resources.Identifier;
 
 import net.fabricmc.fabric.impl.attachment.sync.AttachmentChange;
 
-public record ClientboundAttachmentSyncPayload(List<AttachmentChange> attachments) implements CustomPacketPayload {
+public record ClientboundAttachmentSyncPayload(AttachmentChange attachment) implements CustomPacketPayload {
 	public static final StreamCodec<FriendlyByteBuf, ClientboundAttachmentSyncPayload> CODEC = StreamCodec.composite(
-			AttachmentChange.PACKET_CODEC.apply(ByteBufCodecs.list()), ClientboundAttachmentSyncPayload::attachments,
+			AttachmentChange.PACKET_CODEC,
+			ClientboundAttachmentSyncPayload::attachment,
 			ClientboundAttachmentSyncPayload::new
 	);
 	public static final Identifier PACKET_ID = Identifier.fromNamespaceAndPath("fabric", "attachment_sync_v1");
-	public static final Type<ClientboundAttachmentSyncPayload> ID = new Type<>(PACKET_ID);
+	public static final Type<ClientboundAttachmentSyncPayload> TYPE = new Type<>(PACKET_ID);
 
 	@Override
 	public Type<? extends CustomPacketPayload> type() {
-		return ID;
+		return TYPE;
 	}
 }

fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/mixin/attachment/ServerboundCustomPayloadPacketAccessor.java → fabric-data-attachment-api-v1/src/main/java/net/fabricmc/fabric/mixin/attachment/ClientboundCustomPayloadPacketAccessor.java

@@ -19,10 +19,10 @@
 import org.spongepowered.asm.mixin.Mixin;
 import org.spongepowered.asm.mixin.gen.Accessor;
 
-import net.minecraft.network.protocol.common.ServerboundCustomPayloadPacket;
+import net.minecraft.network.protocol.common.ClientboundCustomPayloadPacket;
 
-@Mixin(ServerboundCustomPayloadPacket.class)
-public interface ServerboundCustomPayloadPacketAccessor {
+@Mixin(ClientboundCustomPayloadPacket.class)
+public interface ClientboundCustomPayloadPacketAccessor {
 	@Accessor("MAX_PAYLOAD_SIZE")
 	static int getMaxPayloadSize() {
 		throw new UnsupportedOperationException("Implemented via mixin");