The server sends and the client stores incorrect observed_transaction_data

[alco]

When the sync service sends SatOpAdditionalBegin and SatOpAdditionalCommit ops, it includes the so-called move_in_ref in their payload. The client stores additional data refs into its local seenAdditionalData meta row, as a comma-separated list of integeres encoded into a string:

electric/clients/typescript/src/satellite/process.ts

Lines 1409 to 1416 in 0115a0a

	async _applyAdditionalData(data: AdditionalData) {
	// Server sends additional data on move-ins and tries to send only data
	// the client has never seen from its perspective. Because of this, we're writing this
	// data directly, like subscription data
	return this._applySubscriptionData(data.changes, this._lsn!, [
	this._addSeenAdditionalDataStmt(data.ref.toString()),
	])
	}

electric/clients/typescript/src/satellite/process.ts

Lines 1460 to 1470 in 0115a0a

	_addSeenAdditionalDataStmt(ref: string): Statement {
	const meta = this.opts.metaTable.toString()
	const sql = `
	INSERT INTO ${meta} (key, value) VALUES ('seenAdditionalData', ?)
	ON CONFLICT (key) DO
	UPDATE SET value = value || ',' || excluded.value
	`
	const args = [ref]
	return { sql, args }
	}

This seenAdditionalData is later fetched, parsed, and included in the SatInStartReplicationReq message's observed_transaction_data field:

electric/clients/typescript/src/satellite/process.ts

Lines 879 to 889 in 0115a0a

	const observedTransactionData = await this._getMeta('seenAdditionalData')
	const { error } = await this.client.startReplication(
	this._lsn,
	schemaVersion,
	subscriptionIds.length > 0 ? subscriptionIds : undefined,
	observedTransactionData
	.split(',')
	.filter((x) => x !== '')
	.map((x) => Long.fromString(x))
	)

electric/clients/typescript/src/satellite/client.ts

Lines 375 to 379 in 0115a0a

	request = SatInStartReplicationReq.fromPartial({
	lsn,
	subscriptionIds,
	observedTransactionData,
	})

When the sync service processes that message, it expects the observed_transaction_data field to contain a list of transaction IDs, though:

electric/components/electric/lib/electric/satellite/protocol.ex

Line 1141 in 0115a0a

including_data: observed_txn_data,

electric/components/electric/lib/electric/satellite/client_reconnection_info.ex

Line 320 in 0115a0a

txids = Keyword.fetch!(opts, :including_data)

[linear]

VAX-1818 The server sends and the client stores incorrect `observed_transaction_data`

[alco]

The same problem is present in the SatOpLogAck message. The protocol definition for this message defines the additional_data_source_ids field:

electric/protocol/satellite.proto

Lines 245 to 246 in 0115a0a

	/** Transaction IDs for which additional data was received immediately after this transaction */
	repeated uint64 additional_data_source_ids = 5;

However, on the client, move_in_refs from SatOpAdditionalCommit messages are used to popoulate this field:

electric/clients/typescript/src/satellite/client.ts

Lines 1057 to 1070 in 0115a0a

	if (op.additionalCommit) {
	if (replication.incomplete !== 'additionalData')
	throw new Error(
	'Unexpected additionalCommit message while not waiting for additionalData'
	)
	const ref = op.additionalCommit!.ref
	// TODO: We need to include these in the ACKs as well
	this.emitter.enqueueEmit(
	'additionalData',
	replication.additionalData[lastDataIdx],
	() => {
	this.inbound.seenAdditionalDataSinceLastTx.dataRefs.push(ref)
	this.maybeSendAck('additionalData')

electric/clients/typescript/src/satellite/client.ts

Lines 1216 to 1217 in 0115a0a

	additionalDataSourceIds:
	this.inbound.seenAdditionalDataSinceLastTx.dataRefs,