internal

package
v0.1.1 Latest Latest
Warning

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

 Go to latest Published: Oct 23, 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 

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 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
}

func (*UserInternalAPI) DeleteKeyBackup 

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

func (*UserInternalAPI) InputAccountData 

func (a *UserInternalAPI) InputAccountData(ctx context.Context, req *api.InputAccountDataRequest, res *api.InputAccountDataResponse) error

func (*UserInternalAPI) PerformAccountCreation 

func (a *UserInternalAPI) PerformAccountCreation(ctx context.Context, req *api.PerformAccountCreationRequest, res *api.PerformAccountCreationResponse) error

func (*UserInternalAPI) PerformAccountDeactivation 

func (a *UserInternalAPI) PerformAccountDeactivation(ctx context.Context, req *api.PerformAccountDeactivationRequest, res *api.PerformAccountDeactivationResponse) error

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 (a *UserInternalAPI) PerformClaimKeys(ctx context.Context, req *api.PerformClaimKeysRequest, res *api.PerformClaimKeysResponse)

func (*UserInternalAPI) PerformDeleteKeys 

func (a *UserInternalAPI) PerformDeleteKeys(ctx context.Context, req *api.PerformDeleteKeysRequest, res *api.PerformDeleteKeysResponse) error

func (*UserInternalAPI) PerformDeviceCreation 

func (a *UserInternalAPI) PerformDeviceCreation(ctx context.Context, req *api.PerformDeviceCreationRequest, res *api.PerformDeviceCreationResponse) error

func (*UserInternalAPI) PerformDeviceDeletion 

func (a *UserInternalAPI) PerformDeviceDeletion(ctx context.Context, req *api.PerformDeviceDeletionRequest, res *api.PerformDeviceDeletionResponse) error

func (*UserInternalAPI) PerformDeviceUpdate 

func (a *UserInternalAPI) PerformDeviceUpdate(ctx context.Context, req *api.PerformDeviceUpdateRequest, res *api.PerformDeviceUpdateResponse) 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 (a *UserInternalAPI) PerformLastSeenUpdate(
	ctx context.Context,
	req *api.PerformLastSeenUpdateRequest,
	res *api.PerformLastSeenUpdateResponse,
) error

func (*UserInternalAPI) PerformLoginTokenCreation 

func (a *UserInternalAPI) PerformLoginTokenCreation(ctx context.Context, req *api.PerformLoginTokenCreationRequest, res *api.PerformLoginTokenCreationResponse) error

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

func (*UserInternalAPI) PerformLoginTokenDeletion 

func (a *UserInternalAPI) PerformLoginTokenDeletion(ctx context.Context, req *api.PerformLoginTokenDeletionRequest, res *api.PerformLoginTokenDeletionResponse) error

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 

func (a *UserInternalAPI) PerformOpenIDTokenCreation(ctx context.Context, req *api.PerformOpenIDTokenCreationRequest, res *api.PerformOpenIDTokenCreationResponse) error

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

func (*UserInternalAPI) PerformPasswordUpdate 

func (a *UserInternalAPI) PerformPasswordUpdate(ctx context.Context, req *api.PerformPasswordUpdateRequest, res *api.PerformPasswordUpdateResponse) error

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 

func (a *UserInternalAPI) PerformUploadDeviceKeys(ctx context.Context, req *api.PerformUploadDeviceKeysRequest, res *api.PerformUploadDeviceKeysResponse)

nolint:gocyclo

func (*UserInternalAPI) PerformUploadDeviceSignatures 

func (a *UserInternalAPI) PerformUploadDeviceSignatures(ctx context.Context, req *api.PerformUploadDeviceSignaturesRequest, res *api.PerformUploadDeviceSignaturesResponse)

func (*UserInternalAPI) PerformUploadKeys 

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

func (*UserInternalAPI) QueryAccessToken 

func (a *UserInternalAPI) QueryAccessToken(ctx context.Context, req *api.QueryAccessTokenRequest, res *api.QueryAccessTokenResponse) error

func (*UserInternalAPI) QueryAccountAvailability 

func (a *UserInternalAPI) QueryAccountAvailability(ctx context.Context, req *api.QueryAccountAvailabilityRequest, res *api.QueryAccountAvailabilityResponse) error

func (*UserInternalAPI) QueryAccountByLocalpart 

func (a *UserInternalAPI) QueryAccountByLocalpart(ctx context.Context, req *api.QueryAccountByLocalpartRequest, res *api.QueryAccountByLocalpartResponse) (err error)

func (*UserInternalAPI) QueryAccountByPassword 

func (a *UserInternalAPI) QueryAccountByPassword(ctx context.Context, req *api.QueryAccountByPasswordRequest, res *api.QueryAccountByPasswordResponse) error

func (*UserInternalAPI) QueryAccountData 

func (a *UserInternalAPI) QueryAccountData(ctx context.Context, req *api.QueryAccountDataRequest, res *api.QueryAccountDataResponse) error

func (*UserInternalAPI) QueryDeviceInfos 

func (a *UserInternalAPI) QueryDeviceInfos(ctx context.Context, req *api.QueryDeviceInfosRequest, res *api.QueryDeviceInfosResponse) error

func (*UserInternalAPI) QueryDeviceMessages 

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

func (*UserInternalAPI) QueryDevices 

func (a *UserInternalAPI) QueryDevices(ctx context.Context, req *api.QueryDevicesRequest, res *api.QueryDevicesResponse) error

func (*UserInternalAPI) QueryKeyBackup 

func (a *UserInternalAPI) QueryKeyBackup(ctx context.Context, req *api.QueryKeyBackupRequest) (*api.QueryKeyBackupResponse, error)

func (*UserInternalAPI) QueryKeyChanges 

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

func (*UserInternalAPI) QueryKeys 

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

nolint:gocyclo

func (*UserInternalAPI) QueryLocalpartForThreePID 

func (a *UserInternalAPI) QueryLocalpartForThreePID(ctx context.Context, req *api.QueryLocalpartForThreePIDRequest, res *api.QueryLocalpartForThreePIDResponse) error

func (*UserInternalAPI) QueryLoginToken 

func (a *UserInternalAPI) QueryLoginToken(ctx context.Context, req *api.QueryLoginTokenRequest, res *api.QueryLoginTokenResponse) error

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 (a *UserInternalAPI) QueryNotifications(ctx context.Context, req *api.QueryNotificationsRequest, res *api.QueryNotificationsResponse) error

func (*UserInternalAPI) QueryNumericLocalpart 

func (a *UserInternalAPI) QueryNumericLocalpart(ctx context.Context, req *api.QueryNumericLocalpartRequest, res *api.QueryNumericLocalpartResponse) error

func (*UserInternalAPI) QueryOneTimeKeys 

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

func (*UserInternalAPI) QueryOpenIDToken 

func (a *UserInternalAPI) QueryOpenIDToken(ctx context.Context, req *api.QueryOpenIDTokenRequest, res *api.QueryOpenIDTokenResponse) error

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 (a *UserInternalAPI) QueryPushers(ctx context.Context, req *api.QueryPushersRequest, res *api.QueryPushersResponse) error

func (*UserInternalAPI) QuerySearchProfiles 

func (a *UserInternalAPI) QuerySearchProfiles(ctx context.Context, req *api.QuerySearchProfilesRequest, res *api.QuerySearchProfilesResponse) error

func (*UserInternalAPI) QuerySignatures 

func (a *UserInternalAPI) QuerySignatures(ctx context.Context, req *api.QuerySignaturesRequest, res *api.QuerySignaturesResponse)

func (*UserInternalAPI) QueryThreePIDsForLocalpart 

func (a *UserInternalAPI) QueryThreePIDsForLocalpart(ctx context.Context, req *api.QueryThreePIDsForLocalpartRequest, res *api.QueryThreePIDsForLocalpartResponse) error

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)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.