Start application service stream token tracking from 1

[anoadragon453]

Warning: this explanation is a bit complex. Let me know if by reading to the end of it it'd be easier to explain over voice.

---

An off-by-one error. Pulled out of #11881, though I felt it deserved a bit more attention than being buried in there.

The application_services_state table is used, among other things, to track the last stream token (5, 6, 7...) of a given stream type (presence, typing, device lists, etc.) that a given application service has previously been informed of. The ApplicationServiceHandler then uses this to determine the set (from the stored stream token up to the latest known stream token) of new presence updates, typing notifications, etc. to send to the application service.

For example, say this is the stream of presence updates that have occurred:

presence update stream tokens:
	2 3 4 5 6 7 8

the application_services_state helps track which events have already been sent out to a given appservice:

presence update stream tokens:
	2 3 4 5 6 7 8
        |
       AS1

with this, the ApplicationServiceHandler knows that it needs to send out the presence updates with stream tokens 4-8 the next time.

I noticed an edge case during #11881, in that the default value for a stream token across Synapse is 1: StreamtokenGenerator sequences start at step (defined as 1 for everything except backfill):

synapse/synapse/storage/util/id_generators.py

Lines 86 to 89 in 0326888

	current_id = int(val) if val else step
	res = (max if step > 0 else min)(current_id, step)
	logger.info("Initialising stream generator for %s(%s): %i", table, column, res)
	return res

MultiWritertokenGenerator sequences also start at 1 (though the first entity will actually start at 2):

synapse/synapse/storage/util/id_generators.py

Lines 344 to 349 in 10a88ba

	# We start at 1 here as a) the first generated stream ID will be 2, and
	# b) other parts of the code assume that stream IDs are strictly greater
	# than 0.
	self._persisted_upto_position = (
	min(self._current_positions.values()) if self._current_positions else 1
	)

As an aside, when we create an event, we usually ask for the next available ID, which is initially 2. So the minimum stream token for any entity is 2.

This is inconsistent with get_type_stream_id_for_appservice, which assumes stream tokens start from 0 by default:

synapse/synapse/storage/databases/main/appservice.py

Lines 448 to 449 in 2cc5ea9

	if last_stream_id is None or last_stream_id[0] is None: # no row exists
	return 0

Why would that be a problem? Well, it turns out some bits of the codebase assume stream token positions start at 1, and checking against 0 can cause bugs. I ran into one here:

synapse/synapse/storage/databases/main/devices.py

Lines 702 to 708 in 53f9d01

	# Get set of users who *may* have changed. Users not in the returned
	# list have definitely not changed.
	if user_ids is None:
	# Get set of all users that have had device list changes since 'from_key'
	user_ids_to_check = self._device_list_stream_cache.get_all_entities_changed(
	from_key
	)

where we call get_all_entities_changed with a minimum stream token to figure out which user tokens have had their device lists changed. This method has a conditional here:

synapse/synapse/util/caches/stream_change_cache.py

Lines 147 to 156 in f8d0f72

	def get_all_entities_changed(self, stream_pos: int) -> Optional[List[EntityType]]:
	"""Returns all entities that have had new things since the given
	position. If the position is too old it will return None.
	Returns the entities in the order that they were changed.
	"""
	assert type(stream_pos) is int
	if stream_pos < self._earliest_known_stream_pos:
	return None

that we kept unintentionally hitting, as self._earliest_known_stream_pos is 1 by default (it is derived from the current stream token when the homeserver starts). Thus this function was returning None, even though there was a device list change (at stream token 2) that had occurred, and that our appservice should know about.

---

The impact of this is that the appservice would never receive word of that first device list change, as from_key would then be incremented.

For any positive impacts on existing EDU -> AS code, I can confirm that presence's logic has a similar bug: from_key would be 0 here:

synapse/synapse/handlers/presence.py

Lines 1630 to 1660 in e24ff8e

	if from_key:
	# First get all users that have had a presence update
	updated_users = stream_change_cache.get_all_entities_changed(from_key)
	# Cross-reference users we're interested in with those that have had updates.
	# Use a slightly-optimised method for processing smaller sets of updates.
	if updated_users is not None and len(updated_users) < 500:
	# For small deltas, it's quicker to get all changes and then
	# cross-reference with the users we're interested in
	get_updates_counter.labels("stream").inc()
	for other_user_id in updated_users:
	if other_user_id in users_interested_in:
	# mypy thinks this variable could be a FrozenSet as it's possibly set
	# to one in the `get_entities_changed` call below, and `add()` is not
	# method on a FrozenSet. That doesn't affect us here though, as
	# `interested_and_updated_users` is clearly a set() above.
	interested_and_updated_users.add(other_user_id) # type: ignore
	else:
	# Too many possible updates. Find all users we can see and check
	# if any of them have changed.
	get_updates_counter.labels("full").inc()
	interested_and_updated_users = (
	stream_change_cache.get_entities_changed(
	users_interested_in, from_key
	)
	)
	else:
	# No from_key has been specified. Return the presence for all users
	# this user is interested in
	interested_and_updated_users = users_interested_in

which looks like it would lead to sending all known presence states to the AS (#10836?).

Typing didn't have this bug as we don't record stream tokens for that, and receipts is not affected either as it just applies the range to the database without a stream cache potentially bailing out in the middle.

[erikjohnston]

Well this is fun, thanks for the explanation! The fix certainly makes sense.

Though I wonder if we actually want to send all the current presence updates to the application service on first start? Otherwise it won't ever see people that are e.g. always online.

It's worth noting that the code has specifically chosen to send all presence updates to the streamer if it falls too far behind. If that is much of a problem then we should also fix that.

[anoadragon453]

@erikjohnston Thanks for taking a look!

By "all known presence states" above, I meant that the AS would be receiving presence updates for every user, not just users the AS is interested in receiving the presence of.

Looking through again though, I think I misread this bit of code as including every user, which it isn't.

synapse/synapse/handlers/presence.py

Lines 1657 to 1660 in e24ff8e

	else:
	# No from_key has been specified. Return the presence for all users
	# this user is interested in
	interested_and_updated_users = users_interested_in

For clarity, we're now going through the truthy body of this conditional, whereas before from_key would be zero, and we'd drop down to the falsey:

synapse/synapse/handlers/presence.py

Lines 1630 to 1660 in e24ff8e

	if from_key:
	# First get all users that have had a presence update
	updated_users = stream_change_cache.get_all_entities_changed(from_key)
	# Cross-reference users we're interested in with those that have had updates.
	# Use a slightly-optimised method for processing smaller sets of updates.
	if updated_users is not None and len(updated_users) < 500:
	# For small deltas, it's quicker to get all changes and then
	# cross-reference with the users we're interested in
	get_updates_counter.labels("stream").inc()
	for other_user_id in updated_users:
	if other_user_id in users_interested_in:
	# mypy thinks this variable could be a FrozenSet as it's possibly set
	# to one in the `get_entities_changed` call below, and `add()` is not
	# method on a FrozenSet. That doesn't affect us here though, as
	# `interested_and_updated_users` is clearly a set() above.
	interested_and_updated_users.add(other_user_id) # type: ignore
	else:
	# Too many possible updates. Find all users we can see and check
	# if any of them have changed.
	get_updates_counter.labels("full").inc()
	interested_and_updated_users = (
	stream_change_cache.get_entities_changed(
	users_interested_in, from_key
	)
	)
	else:
	# No from_key has been specified. Return the presence for all users
	# this user is interested in
	interested_and_updated_users = users_interested_in

So actually, this change is removing presence updates from the initial burst for users that have never set a presence state. Instead of receiving an "offline" state for them, they'll just be omitted. This is a change in behaviour, but probably isn't a problem as the AS would just assume those users would be offline anyways?

...I'm going to change that to if from_key is not None here as well to avoid it catching 0 at some other point.