[Kafka] Raise the producer message cap so policy-max transactions publish

[Calgooon]

What Changed

• kafka/sarama_broker.go: new DefaultProducerMaxMessageBytes (16 MiB) and NewSaramaBrokerWithMaxMessageBytes; newSyncProducerConfig takes the cap and sets Producer.MaxMessageBytes on both the sync and async producers. NewSaramaBroker / NewSaramaBrokerWithLogger keep their signatures and use the default.
• kafka/factory.go: passes cfg.MaxMessageBytes through.
• config/config.go + config.example.yaml: new kafka.max_message_bytes (0 = default).
• compose/topic-init.sh: sets kafka_batch_max_bytes and per-topic max.message.bytes (create and idempotent alter for pre-existing topics), env-overridable via MAX_MESSAGE_BYTES.
• deploy/kafka.yaml: KAFKA_CFG_MESSAGE_MAX_BYTES and KAFKA_CFG_REPLICA_FETCH_MAX_BYTES = 16 MiB.
• docs/production-kafka.md: "Message size" section listing the three limits that must agree; notes the bsva-infra-flux topic config.
• kafka/sarama_broker_test.go: TestSyncProducerConfig_MaxMessageBytes pins the default (and that it clears a base64-encoded 10 MiB tx) and the passthrough; existing callers of newSyncProducerConfig updated.

Why It Was Necessary

The producers ran on sarama's default 1 MiB cap. A TopicPropagation message carries raw_tx base64-encoded inside JSON (x4/3), so transactions above ~750 KB of EF failed at producer.Send with 500 failed to submit even though every advertised policy limit (10 MiB tx, network-tracked script size) accepted them. bump_builder already works around the same default by chunking; a single transaction cannot be chunked. Fixes #329.

Testing Performed

• gofmt -l kafka/ config/ clean; go build ./kafka/ ./config/, go vet ./kafka/ ./config/, go test ./kafka/ pass (includes the new test).
• Not run here: make lint (golangci-lint not installed on this machine) and the compose stack; topic-init.sh changes follow the existing rpk usage in the script and are idempotent.

Impact / Risk

• Producer-side cap only becomes effective when the broker and topic accept the same size; the compose and deploy manifests are updated, and the doc names the bsva-infra-flux change production needs. Without the broker side, behaviour is unchanged (produce still fails broker-side with the same 500).
• Memory: a 16 MiB message is buffered per submit; the API already caps single-tx bodies at 32 MiB, so this is not a new ceiling.
• No API or storage changes; NewSaramaBroker callers (e2e tests) are unaffected.

Notifications

• @CrosTheRubicon

[github-actions]

👋 Thanks, @Calgooon!

This pull request comes from a fork. For security, our CI runs in a restricted mode.
A maintainer will triage this shortly and run any additional checks as needed.

• 🏷️ Labeled: fork-pr, requires-manual-review
• 👀 We'll review and follow up here if anything else is needed.

Thanks for contributing to bsv-blockchain/arcade! 🚀

[galt-tr]

Issue #329: valid. On c49c8dd neither producer sets Producer.MaxMessageBytes, so sarama v1.60.1's 1 MiB default applies (config.go:569); the check runs in the async dispatcher on the uncompressed ByteSize (async_producer.go:653-655) and SyncProducer wraps that producer, so the base64 raw_tx in the propagation envelope (services/api_server/handlers.go:952-958) trips it at ~750 KB of tx and maps to 500 {"error":"failed to submit"} (handlers.go:967-971). services/bump_builder/builder.go:410-415 already chunks bulk MINED events around the same default.

Overall the change is small and correctly scoped: both producers get the cap, NewBroker threads cfg.MaxMessageBytes through, NewSaramaBroker/WithLogger keep their signatures, and the compose and dev-StatefulSet broker/topic sides move in the same PR. gofmt, go build ./..., go vet (incl. -tags=e2e ./tests/e2e/), go test ./kafka/... ./config/... and golangci-lint are clean locally (CI's Test Suite job does not run for fork PRs). Two things need fixing before merge: the new key has no viper.SetDefault, so ARCADE_KAFKA_MAX_MESSAGE_BYTES is silently ignored (the only way deploy/ configures kafka); and the 16 MiB default does not cover a policy-max transaction once input_txids and EF per-input data are counted (measured ~21 MiB for a 9.99 MiB P2PKH consolidation submitted as EF), so the const comment, test and docs overstate what the default guarantees. Details inline.

Not in the diff:

• services/bump_builder/builder.go:410-415 and :458-462 still cite "the 1 MiB default Producer.MaxMessageBytes" as the reason for maxTxIDsPerBulkEvent = 5000. The chunking should stay (an operator can set kafka.max_message_bytes below the new default), but the comments should point at kafka.max_message_bytes / kafka.DefaultProducerMaxMessageBytes instead of a default that no longer applies.

Checked and not an issue: the consumer can read 16 MiB records (newSaramaConsumerConfig leaves Consumer.Fetch.Max at 0, so the ErrMessageTooLarge branch at consumer.go:917-919 is unreachable and the fetch size grows to the partial batch size; Fetch.MaxBytes defaults to 50 MiB); the async producer's unset Version resolves to sarama's DefaultVersion (2.8.0), so both producers size messages with record-v2 overhead; per-partition batches now roll over at 16 MiB instead of 1 MiB (produce_set.go:327-328), which matches the broker/topic caps set here and does not add memory since queued messages are already resident; rpk topic create -c/--topic-config, rpk topic alter-config --set, rpk cluster config set kafka_batch_max_bytes and the bitnami KAFKA_CFG_* mappings match the documented CLI/env forms (not executed here).

[galt-tr]

Important kafka.max_message_bytes has no viper.SetDefault in setDefaults(), while every other kafka.* key does (L1108-1113). Load uses AutomaticEnv + viper.Unmarshal, and Unmarshal only visits keys viper already knows from defaults/config/flags, so ARCADE_KAFKA_MAX_MESSAGE_BYTES is silently dropped whenever the key is absent from the YAML. Verified against viper v1.21.0 with this exact prefix/replacer setup: viper.GetInt("kafka.max_message_bytes") returns the env value but the unmarshalled field stays 0, and adding the default fixes it. deploy/*.yaml configure kafka exclusively through ARCADE_KAFKA_* env, so as written the cap cannot be changed there at all, and an operator who raises it will keep getting the same 500 failed to submit (the comment at L1300-1301 documents this requirement for the validator keys).

	viper.SetDefault("kafka.send_timeout_ms", 2000)
	viper.SetDefault("kafka.max_message_bytes", 0)

[galt-tr]

Important The x4/3 derivation counts only raw_tx. The propagation message also carries input_txids, one 64-hex string per input with no dedup (~67 B each, services/api_server/handlers.go:37-49,952-956), and raw_tx is whatever bytes the client sent, so an EF submit adds ~34 B per P2PKH input before base64. The size policy is enforced on the non-EF serialisation (tx.Size() via gobdk SetMaxTxSizePolicy), so a consolidation-shaped tx is policy-valid well past this cap. Measured with go-sdk v1.3.3 (P2PKH inputs, one output, the same map the handler marshals):

inputs	tx.Size()	message, raw submit	message, EF submit
55,000	7.76 MiB	13.86 MiB	16.24 MiB
60,000	8.47 MiB	15.13 MiB	17.72 MiB
70,800	9.99 MiB	17.85 MiB	20.91 MiB

With this default an EF consolidation above ~7.5 MiB, or a raw one above ~9 MiB, still fails with the deterministic 500 the issue describes, while this comment, the new test and docs/production-kafka.md say it is covered. Either raise the default (32 MiB, matching maxSingleTxBytes, clears the policy-max P2PKH case with margin; compose/topic-init.sh and deploy/kafka.yaml need the same value) or state the real bound here: base64 of the submitted bytes plus ~67 B per input. The bound is O(inputs), so the doc should say which tx shapes the chosen default covers rather than "policy-max".

[galt-tr]

Minor Nothing rejects a configured value at or above sarama.MaxRequestSize (100 MiB). sarama's Config.Validate only logs Producer.MaxMessageBytes must be smaller than MaxRequestSize; it will be ignored (config.go:639-640) and the dispatcher keeps using the value, so the mistake surfaces per message at produce time (wouldOverflow on MaxRequestSize-10KiB) instead of at startup. Fail fast here:

	if maxMessageBytes >= int(sarama.MaxRequestSize) {
		return nil, fmt.Errorf("kafka.max_message_bytes %d must be below sarama.MaxRequestSize (%d)", maxMessageBytes, sarama.MaxRequestSize)
	}

[galt-tr]

Minor minNeeded models only base64 of a non-EF 10 MiB tx: 10485760*4/3 is 13,981,013 B, which is 13.3 MiB (14.0 MB), so the "~14 MiB" wording here, in the const doc and in the docs mixes units. It omits the input_txids array and EF per-input data, which is what pushes a policy-max consolidation to ~21 MiB (see the comment on DefaultProducerMaxMessageBytes), so this assertion passes while the guarantee it describes does not hold. If the default is meant to guarantee "policy-valid submit fits", pin that with a synthetic tx (N P2PKH inputs via go-sdk with tx.Size() just under the policy, marshal the same map[string]interface{} the handler builds, assert len(b) < DefaultProducerMaxMessageBytes) so the test moves when the envelope does. The explicit-value case should also assert Validate() passes, as TestNewSyncProducerConfig_OrderingInvariants does for the default.

[galt-tr]

Minor compose/arcade-config.yaml has no kafka.max_message_bytes (its kafka: block only sets brokers), so this pointer sends the reader to a key that is not there; the documented key is in config.example.yaml. Because the compose arcade instance therefore always runs the code default, an operator who overrides MAX_MESSAGE_BYTES here changes the broker and topic caps but not the producer. Either add max_message_bytes: 16777216 to compose/arcade-config.yaml and keep this reference, or point it at config.example.yaml / kafka.DefaultProducerMaxMessageBytes.

[galt-tr]

Minor "~14 MiB" is the base64 of a non-EF 10 MiB tx only (13.3 MiB; 14 is the MB figure). The envelope also carries input_txids (~67 B per input, not deduplicated) and an EF submit adds per-input source data, so a policy-max P2PKH consolidation submitted as EF measures ~21 MiB, above the 16 MiB the table below presents as sufficient. State the real bound (base64 of the submitted bytes + ~67 B x inputs) and which tx shapes the default covers, or raise the default together with the two broker values.