internal

Documentation

Index

• type DeviceListUpdater
  • func NewDeviceListUpdater(db DeviceListUpdaterDatabase, producer KeyChangeProducer, ...) *DeviceListUpdater
  • func (u *DeviceListUpdater) ManualUpdate(ctx context.Context, serverName gomatrixserverlib.ServerName, userID string) error
  • func (u *DeviceListUpdater) Start() error
  • func (u *DeviceListUpdater) Update(ctx context.Context, event gomatrixserverlib.DeviceListUpdateEvent) error
• type DeviceListUpdaterDatabase
• type KeyChangeProducer
• type KeyInternalAPI
  • func (a *KeyInternalAPI) InputDeviceListUpdate(ctx context.Context, req *api.InputDeviceListUpdateRequest, ...)
  • func (a *KeyInternalAPI) PerformClaimKeys(ctx context.Context, req *api.PerformClaimKeysRequest, ...)
  • func (a *KeyInternalAPI) PerformUploadKeys(ctx context.Context, req *api.PerformUploadKeysRequest, ...)
  • func (a *KeyInternalAPI) QueryDeviceMessages(ctx context.Context, req *api.QueryDeviceMessagesRequest, ...)
  • func (a *KeyInternalAPI) QueryKeyChanges(ctx context.Context, req *api.QueryKeyChangesRequest, ...)
  • func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysRequest, res *api.QueryKeysResponse)
  • func (a *KeyInternalAPI) QueryOneTimeKeys(ctx context.Context, req *api.QueryOneTimeKeysRequest, ...)
  • func (a *KeyInternalAPI) SetUserAPI(i userapi.UserInternalAPI)

Types

type DeviceListUpdater

type DeviceListUpdater struct {
	// contains filtered or unexported fields
}

DeviceListUpdater handles device list updates from remote servers.

In the case where we have the prev_id for an update, the updater just stores the update (after acquiring a per-user lock). In the case where we do not have the prev_id for an update, the updater marks the user_id as stale and notifies a worker to get the latest device list for this user. Note: stream IDs are scoped per user so missing a prev_id for a (user, device) does not mean that DEVICE is outdated as the previous ID could be for a different device: we have to invalidate all devices for that user. Once the list has been fetched, the per-user lock is acquired and the updater stores the latest list along with the latest stream ID.

On startup, the updater spins up N workers which are responsible for querying device keys from remote servers. Workers are scoped by homeserver domain, with one worker responsible for many domains, determined by hashing mod N the server name. Work is sent via a channel which just serves to "poke" the worker as the data is retrieved from the database (which allows us to batch requests to the same server). This has a number of desirable properties:

• We guarantee only 1 in-flight /keys/query request per server at any time as there is exactly 1 worker responsible for that domain.
• We don't have unbounded growth in proportion to the number of servers (this is more important in a P2P world where we have many many servers)
• We can adjust concurrency (at the cost of memory usage) by tuning N, to accommodate mobile devices vs servers.

The downsides are that:

• Query requests can get queued behind other servers if they hash to the same worker, even if there are other free workers elsewhere. Whilst suboptimal, provided we cap how long a single request can last (e.g using context timeouts) we guarantee we will get around to it. Also, more users on a given server does not increase the number of requests (as /keys/query allows multiple users to be specified) so being stuck behind matrix.org won't materially be any worse than being stuck behind foo.bar

In the event that the query fails, the worker spins up a short-lived goroutine whose sole purpose is to inject the server name back into the channel after a certain amount of time. If in the interim the device lists have been updated, then the database query will return no stale lists. Reinjection into the channel continues until success or the server terminates, when it will be reloaded on startup.

func NewDeviceListUpdater

func NewDeviceListUpdater(
	db DeviceListUpdaterDatabase, producer KeyChangeProducer, fedClient fedsenderapi.FederationClient,
	numWorkers int,
) *DeviceListUpdater

NewDeviceListUpdater creates a new updater which fetches fresh device lists when they go stale.

func (*DeviceListUpdater) ManualUpdate

func (u *DeviceListUpdater) ManualUpdate(ctx context.Context, serverName gomatrixserverlib.ServerName, userID string) error

ManualUpdate invalidates the device list for the given user and fetches the latest and tracks it. Blocks until the device list is synced or the timeout is reached.

func (*DeviceListUpdater) Start

func (u *DeviceListUpdater) Start() error

Start the device list updater, which will try to refresh any stale device lists.

func (*DeviceListUpdater) Update

func (u *DeviceListUpdater) Update(ctx context.Context, event gomatrixserverlib.DeviceListUpdateEvent) error

Update blocks until the update has been stored in the database. It blocks primarily for satisfying sytest, which assumes when /send 200 OKs that the device lists have been updated.

type DeviceListUpdaterDatabase

type DeviceListUpdaterDatabase interface {
	// StaleDeviceLists returns a list of user IDs ending with the domains provided who have stale device lists.
	// If no domains are given, all user IDs with stale device lists are returned.
	StaleDeviceLists(ctx context.Context, domains []gomatrixserverlib.ServerName) ([]string, error)

	// MarkDeviceListStale sets the stale bit for this user to isStale.
	MarkDeviceListStale(ctx context.Context, userID string, isStale bool) error

	// StoreRemoteDeviceKeys persists the given keys. Keys with the same user ID and device ID will be replaced. An empty KeyJSON removes the key
	// for this (user, device). Does not modify the stream ID for keys. User IDs in `clearUserIDs` will have all their device keys deleted prior
	// to insertion - use this when you have a complete snapshot of a user's keys in order to track device deletions correctly.
	StoreRemoteDeviceKeys(ctx context.Context, keys []api.DeviceMessage, clearUserIDs []string) error

	// PrevIDsExists returns true if all prev IDs exist for this user.
	PrevIDsExists(ctx context.Context, userID string, prevIDs []int) (bool, error)

	// DeviceKeysJSON populates the KeyJSON for the given keys. If any proided `keys` have a `KeyJSON` or `StreamID` already then it will be replaced.
	DeviceKeysJSON(ctx context.Context, keys []api.DeviceMessage) error
}

DeviceListUpdaterDatabase is the subset of functionality from storage.Database required for the updater. Useful for testing.

type KeyChangeProducer

type KeyChangeProducer interface {
	ProduceKeyChanges(keys []api.DeviceMessage) error
}

KeyChangeProducer is the interface for producers.KeyChange useful for testing.

type KeyInternalAPI

type KeyInternalAPI struct {
	DB         storage.Database
	ThisServer gomatrixserverlib.ServerName
	FedClient  fedsenderapi.FederationClient
	UserAPI    userapi.UserInternalAPI
	Producer   *producers.KeyChange
	Updater    *DeviceListUpdater
}

func (*KeyInternalAPI) InputDeviceListUpdate

func (a *KeyInternalAPI) InputDeviceListUpdate(
	ctx context.Context, req *api.InputDeviceListUpdateRequest, res *api.InputDeviceListUpdateResponse,
)

func (*KeyInternalAPI) PerformClaimKeys

func (a *KeyInternalAPI) PerformClaimKeys(ctx context.Context, req *api.PerformClaimKeysRequest, res *api.PerformClaimKeysResponse)

func (*KeyInternalAPI) PerformUploadKeys

func (a *KeyInternalAPI) PerformUploadKeys(ctx context.Context, req *api.PerformUploadKeysRequest, res *api.PerformUploadKeysResponse)

func (*KeyInternalAPI) QueryDeviceMessages

func (a *KeyInternalAPI) QueryDeviceMessages(ctx context.Context, req *api.QueryDeviceMessagesRequest, res *api.QueryDeviceMessagesResponse)

func (*KeyInternalAPI) QueryKeyChanges

func (a *KeyInternalAPI) QueryKeyChanges(ctx context.Context, req *api.QueryKeyChangesRequest, res *api.QueryKeyChangesResponse)

func (*KeyInternalAPI) QueryKeys

func (a *KeyInternalAPI) QueryKeys(ctx context.Context, req *api.QueryKeysRequest, res *api.QueryKeysResponse)

func (*KeyInternalAPI) QueryOneTimeKeys

func (a *KeyInternalAPI) QueryOneTimeKeys(ctx context.Context, req *api.QueryOneTimeKeysRequest, res *api.QueryOneTimeKeysResponse)

func (*KeyInternalAPI) SetUserAPI

func (a *KeyInternalAPI) SetUserAPI(i userapi.UserInternalAPI)