internal

package
v0.2.1 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Nov 6, 2024 License: Apache-2.0 Imports: 44 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrIsRemoteServer = errors.New("cannot query profile of remote users")
)

Functions

This section is empty.

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, a lock is acquired and the server name along with the time to wait before retrying is set in a map. A restarter goroutine periodically probes this map and injects servers which are ready to be retried.

func NewDeviceListUpdater

func NewDeviceListUpdater(
	process *process.ProcessContext, db DeviceListUpdaterDatabase,
	api DeviceListUpdaterAPI, producer KeyChangeProducer,
	fedClient fedsenderapi.KeyserverFederationAPI, numWorkers int,
	rsAPI rsapi.KeyserverRoomserverAPI,
	thisServer spec.ServerName,
	enableMetrics bool,
	isBlacklistedOrBackingOffFn func(s spec.ServerName) (*statistics.ServerStatistics, error),
) *DeviceListUpdater

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

func (*DeviceListUpdater) CleanUp

func (u *DeviceListUpdater) CleanUp() error

CleanUp removes stale device entries for users we don't share a room with anymore

func (*DeviceListUpdater) ManualUpdate

func (u *DeviceListUpdater) ManualUpdate(ctx context.Context, serverName spec.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

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 DeviceListUpdaterAPI

type DeviceListUpdaterAPI interface {
	PerformUploadDeviceKeys(ctx context.Context, req *api.PerformUploadDeviceKeysRequest, res *api.PerformUploadDeviceKeysResponse)
}

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 []spec.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 []int64) (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

	DeleteStaleDeviceLists(ctx context.Context, userIDs []string) 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 UserInternalAPI

type UserInternalAPI struct {
	DB                storage.UserDatabase
	KeyDatabase       storage.KeyDatabase
	SyncProducer      *producers.SyncAPI
	KeyChangeProducer *producers.KeyChange
	Config            *config.UserAPI

	DisableTLSValidation bool
	// AppServices is the list of all registered AS
	AppServices []config.ApplicationService
	RSAPI       rsapi.UserRoomserverAPI
	PgClient    pushgateway.Client
	FedClient   fedsenderapi.KeyserverFederationAPI
	Updater     *DeviceListUpdater

	JWTCertMap map[string]string
}

func (*UserInternalAPI) DeleteKeyBackup

func (a *UserInternalAPI) DeleteKeyBackup(ctx context.Context, userID, version string) (bool, error)

func (*UserInternalAPI) InputAccountData

func (*UserInternalAPI) PerformAccountCreation

func (*UserInternalAPI) PerformAccountDeactivation

PerformAccountDeactivation deactivates the user's account, removing all ability for the user to login again.

func (*UserInternalAPI) PerformAdminCreateRegistrationToken

func (a *UserInternalAPI) PerformAdminCreateRegistrationToken(ctx context.Context, registrationToken *clientapi.RegistrationToken) (bool, error)

func (*UserInternalAPI) PerformAdminDeleteRegistrationToken

func (a *UserInternalAPI) PerformAdminDeleteRegistrationToken(ctx context.Context, tokenString string) error

func (*UserInternalAPI) PerformAdminGetRegistrationToken

func (a *UserInternalAPI) PerformAdminGetRegistrationToken(ctx context.Context, tokenString string) (*clientapi.RegistrationToken, error)

func (*UserInternalAPI) PerformAdminListRegistrationTokens

func (a *UserInternalAPI) PerformAdminListRegistrationTokens(ctx context.Context, returnAll bool, valid bool) ([]clientapi.RegistrationToken, error)

func (*UserInternalAPI) PerformAdminUpdateRegistrationToken

func (a *UserInternalAPI) PerformAdminUpdateRegistrationToken(ctx context.Context, tokenString string, newAttributes map[string]interface{}) (*clientapi.RegistrationToken, error)

func (*UserInternalAPI) PerformClaimKeys

func (*UserInternalAPI) PerformDeleteKeys

func (*UserInternalAPI) PerformDeviceCreation

func (*UserInternalAPI) PerformDeviceDeletion

func (*UserInternalAPI) PerformDeviceUpdate

func (*UserInternalAPI) PerformEnsureSSOAccountExists added in v0.2.1

func (a *UserInternalAPI) PerformEnsureSSOAccountExists(ctx context.Context, req *api.QuerySSOAccountRequest, res *api.QuerySSOAccountResponse) error

func (*UserInternalAPI) PerformForgetThreePID

func (a *UserInternalAPI) PerformForgetThreePID(ctx context.Context, req *api.PerformForgetThreePIDRequest, res *struct{}) error

func (*UserInternalAPI) PerformKeyBackup

func (a *UserInternalAPI) PerformKeyBackup(ctx context.Context, req *api.PerformKeyBackupRequest) (string, error)

func (*UserInternalAPI) PerformLastSeenUpdate

func (*UserInternalAPI) PerformLoginTokenCreation

PerformLoginTokenCreation creates a new login token and associates it with the provided data.

func (*UserInternalAPI) PerformLoginTokenDeletion

PerformLoginTokenDeletion ensures the token doesn't exist.

func (*UserInternalAPI) PerformMarkAsStaleIfNeeded

func (a *UserInternalAPI) PerformMarkAsStaleIfNeeded(ctx context.Context, req *api.PerformMarkAsStaleRequest, res *struct{}) error

PerformMarkAsStaleIfNeeded marks the users device list as stale, if the given deviceID is not present in our database.

func (*UserInternalAPI) PerformOpenIDTokenCreation

PerformOpenIDTokenCreation creates a new token that a relying party uses to authenticate a user

func (*UserInternalAPI) PerformPasswordUpdate

func (*UserInternalAPI) PerformPushRulesPut

func (a *UserInternalAPI) PerformPushRulesPut(
	ctx context.Context,
	userID string,
	ruleSets *pushrules.AccountRuleSets,
) error

func (*UserInternalAPI) PerformPusherDeletion

func (a *UserInternalAPI) PerformPusherDeletion(ctx context.Context, req *api.PerformPusherDeletionRequest, res *struct{}) error

func (*UserInternalAPI) PerformPusherSet

func (a *UserInternalAPI) PerformPusherSet(ctx context.Context, req *api.PerformPusherSetRequest, res *struct{}) error

func (*UserInternalAPI) PerformSaveThreePIDAssociation

func (a *UserInternalAPI) PerformSaveThreePIDAssociation(ctx context.Context, req *api.PerformSaveThreePIDAssociationRequest, res *struct{}) error

func (*UserInternalAPI) PerformUploadDeviceKeys

nolint:gocyclo

func (*UserInternalAPI) PerformUploadKeys

func (*UserInternalAPI) QueryAccessToken

func (*UserInternalAPI) QueryAccountAvailability

func (*UserInternalAPI) QueryAccountByLocalpart

func (*UserInternalAPI) QueryAccountByPassword

func (*UserInternalAPI) QueryAccountData

func (*UserInternalAPI) QueryDeviceInfos

func (*UserInternalAPI) QueryDeviceMessages

func (*UserInternalAPI) QueryDevices

func (*UserInternalAPI) QueryKeyBackup

func (*UserInternalAPI) QueryKeyChanges

func (*UserInternalAPI) QueryKeys

nolint:gocyclo

func (*UserInternalAPI) QueryLocalpartForThreePID

func (*UserInternalAPI) QueryLoginToken

QueryLoginToken returns the data associated with a login token. If the token is not valid, success is returned, but res.Data == nil.

func (*UserInternalAPI) QueryNotifications

func (*UserInternalAPI) QueryNumericLocalpart

func (*UserInternalAPI) QueryOneTimeKeys

func (*UserInternalAPI) QueryOpenIDToken

QueryOpenIDToken validates that the OpenID token was issued for the user, the replying party uses this for validation

func (*UserInternalAPI) QueryProfile

func (a *UserInternalAPI) QueryProfile(ctx context.Context, userID string) (*authtypes.Profile, error)

func (*UserInternalAPI) QueryPushRules

func (a *UserInternalAPI) QueryPushRules(ctx context.Context, userID string) (*pushrules.AccountRuleSets, error)

func (*UserInternalAPI) QueryPushers

func (*UserInternalAPI) QuerySearchProfiles

func (*UserInternalAPI) QuerySignatures

func (*UserInternalAPI) SetAvatarURL

func (a *UserInternalAPI) SetAvatarURL(ctx context.Context, localpart string, serverName spec.ServerName, avatarURL string) (*authtypes.Profile, bool, error)

func (*UserInternalAPI) SetDisplayName

func (a *UserInternalAPI) SetDisplayName(ctx context.Context, localpart string, serverName spec.ServerName, displayName string) (*authtypes.Profile, bool, error)

func (*UserInternalAPI) UpdateBackupKeyAuthData

func (a *UserInternalAPI) UpdateBackupKeyAuthData(ctx context.Context, req *api.PerformKeyBackupRequest) (*api.PerformKeyBackupResponse, error)

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL