import { MongoCompatibilityError, MongoInvalidArgumentError } from '../error'; import { ReadPreference } from '../read_preference'; import { ServerType, TopologyType } from './common'; import type { ServerDescription, TagSet } from './server_description'; import type { TopologyDescription } from './topology_description'; // max staleness constants const IDLE_WRITE_PERIOD = 10000; const SMALLEST_MAX_STALENESS_SECONDS = 90; // Minimum version to try writes on secondaries. export const MIN_SECONDARY_WRITE_WIRE_VERSION = 13; /** @public */ export type ServerSelector = ( topologyDescription: TopologyDescription, servers: ServerDescription[] ) => ServerDescription[]; /** * Returns a server selector that selects for writable servers */ export function writableServerSelector(): ServerSelector { return ( topologyDescription: TopologyDescription, servers: ServerDescription[] ): ServerDescription[] => latencyWindowReducer( topologyDescription, servers.filter((s: ServerDescription) => s.isWritable) ); } /** * The purpose of this selector is to select the same server, only * if it is in a state that it can have commands sent to it. */ export function sameServerSelector(description?: ServerDescription): ServerSelector { return ( topologyDescription: TopologyDescription, servers: ServerDescription[] ): ServerDescription[] => { if (!description) return []; // Filter the servers to match the provided description only if // the type is not unknown. return servers.filter(sd => { return sd.address === description.address && sd.type !== ServerType.Unknown; }); }; } /** * Returns a server selector that uses a read preference to select a * server potentially for a write on a secondary. */ export function secondaryWritableServerSelector( wireVersion?: number, readPreference?: ReadPreference ): ServerSelector { // If server version < 5.0, read preference always primary. // If server version >= 5.0... // - If read preference is supplied, use that. // - If no read preference is supplied, use primary. if ( !readPreference || !wireVersion || (wireVersion && wireVersion < MIN_SECONDARY_WRITE_WIRE_VERSION) ) { return readPreferenceServerSelector(ReadPreference.primary); } return readPreferenceServerSelector(readPreference); } /** * Reduces the passed in array of servers by the rules of the "Max Staleness" specification * found here: https://github.com/mongodb/specifications/blob/master/source/max-staleness/max-staleness.rst * * @param readPreference - The read preference providing max staleness guidance * @param topologyDescription - The topology description * @param servers - The list of server descriptions to be reduced * @returns The list of servers that satisfy the requirements of max staleness */ function maxStalenessReducer( readPreference: ReadPreference, topologyDescription: TopologyDescription, servers: ServerDescription[] ): ServerDescription[] { if (readPreference.maxStalenessSeconds == null || readPreference.maxStalenessSeconds < 0) { return servers; } const maxStaleness = readPreference.maxStalenessSeconds; const maxStalenessVariance = (topologyDescription.heartbeatFrequencyMS + IDLE_WRITE_PERIOD) / 1000; if (maxStaleness < maxStalenessVariance) { throw new MongoInvalidArgumentError( `Option "maxStalenessSeconds" must be at least ${maxStalenessVariance} seconds` ); } if (maxStaleness < SMALLEST_MAX_STALENESS_SECONDS) { throw new MongoInvalidArgumentError( `Option "maxStalenessSeconds" must be at least ${SMALLEST_MAX_STALENESS_SECONDS} seconds` ); } if (topologyDescription.type === TopologyType.ReplicaSetWithPrimary) { const primary: ServerDescription = Array.from(topologyDescription.servers.values()).filter( primaryFilter )[0]; return servers.reduce((result: ServerDescription[], server: ServerDescription) => { const stalenessMS = server.lastUpdateTime - server.lastWriteDate - (primary.lastUpdateTime - primary.lastWriteDate) + topologyDescription.heartbeatFrequencyMS; const staleness = stalenessMS / 1000; const maxStalenessSeconds = readPreference.maxStalenessSeconds ?? 0; if (staleness <= maxStalenessSeconds) { result.push(server); } return result; }, []); } if (topologyDescription.type === TopologyType.ReplicaSetNoPrimary) { if (servers.length === 0) { return servers; } const sMax = servers.reduce((max: ServerDescription, s: ServerDescription) => s.lastWriteDate > max.lastWriteDate ? s : max ); return servers.reduce((result: ServerDescription[], server: ServerDescription) => { const stalenessMS = sMax.lastWriteDate - server.lastWriteDate + topologyDescription.heartbeatFrequencyMS; const staleness = stalenessMS / 1000; const maxStalenessSeconds = readPreference.maxStalenessSeconds ?? 0; if (staleness <= maxStalenessSeconds) { result.push(server); } return result; }, []); } return servers; } /** * Determines whether a server's tags match a given set of tags * * @param tagSet - The requested tag set to match * @param serverTags - The server's tags */ function tagSetMatch(tagSet: TagSet, serverTags: TagSet) { const keys = Object.keys(tagSet); const serverTagKeys = Object.keys(serverTags); for (let i = 0; i < keys.length; ++i) { const key = keys[i]; if (serverTagKeys.indexOf(key) === -1 || serverTags[key] !== tagSet[key]) { return false; } } return true; } /** * Reduces a set of server descriptions based on tags requested by the read preference * * @param readPreference - The read preference providing the requested tags * @param servers - The list of server descriptions to reduce * @returns The list of servers matching the requested tags */ function tagSetReducer( readPreference: ReadPreference, servers: ServerDescription[] ): ServerDescription[] { if ( readPreference.tags == null || (Array.isArray(readPreference.tags) && readPreference.tags.length === 0) ) { return servers; } for (let i = 0; i < readPreference.tags.length; ++i) { const tagSet = readPreference.tags[i]; const serversMatchingTagset = servers.reduce( (matched: ServerDescription[], server: ServerDescription) => { if (tagSetMatch(tagSet, server.tags)) matched.push(server); return matched; }, [] ); if (serversMatchingTagset.length) { return serversMatchingTagset; } } return []; } /** * Reduces a list of servers to ensure they fall within an acceptable latency window. This is * further specified in the "Server Selection" specification, found here: * https://github.com/mongodb/specifications/blob/master/source/server-selection/server-selection.rst * * @param topologyDescription - The topology description * @param servers - The list of servers to reduce * @returns The servers which fall within an acceptable latency window */ function latencyWindowReducer( topologyDescription: TopologyDescription, servers: ServerDescription[] ): ServerDescription[] { const low = servers.reduce( (min: number, server: ServerDescription) => min === -1 ? server.roundTripTime : Math.min(server.roundTripTime, min), -1 ); const high = low + topologyDescription.localThresholdMS; return servers.reduce((result: ServerDescription[], server: ServerDescription) => { if (server.roundTripTime <= high && server.roundTripTime >= low) result.push(server); return result; }, []); } // filters function primaryFilter(server: ServerDescription): boolean { return server.type === ServerType.RSPrimary; } function secondaryFilter(server: ServerDescription): boolean { return server.type === ServerType.RSSecondary; } function nearestFilter(server: ServerDescription): boolean { return server.type === ServerType.RSSecondary || server.type === ServerType.RSPrimary; } function knownFilter(server: ServerDescription): boolean { return server.type !== ServerType.Unknown; } function loadBalancerFilter(server: ServerDescription): boolean { return server.type === ServerType.LoadBalancer; } /** * Returns a function which selects servers based on a provided read preference * * @param readPreference - The read preference to select with */ export function readPreferenceServerSelector(readPreference: ReadPreference): ServerSelector { if (!readPreference.isValid()) { throw new MongoInvalidArgumentError('Invalid read preference specified'); } return ( topologyDescription: TopologyDescription, servers: ServerDescription[] ): ServerDescription[] => { const commonWireVersion = topologyDescription.commonWireVersion; if ( commonWireVersion && readPreference.minWireVersion && readPreference.minWireVersion > commonWireVersion ) { throw new MongoCompatibilityError( `Minimum wire version '${readPreference.minWireVersion}' required, but found '${commonWireVersion}'` ); } if (topologyDescription.type === TopologyType.LoadBalanced) { return servers.filter(loadBalancerFilter); } if (topologyDescription.type === TopologyType.Unknown) { return []; } if ( topologyDescription.type === TopologyType.Single || topologyDescription.type === TopologyType.Sharded ) { return latencyWindowReducer(topologyDescription, servers.filter(knownFilter)); } const mode = readPreference.mode; if (mode === ReadPreference.PRIMARY) { return servers.filter(primaryFilter); } if (mode === ReadPreference.PRIMARY_PREFERRED) { const result = servers.filter(primaryFilter); if (result.length) { return result; } } const filter = mode === ReadPreference.NEAREST ? nearestFilter : secondaryFilter; const selectedServers = latencyWindowReducer( topologyDescription, tagSetReducer( readPreference, maxStalenessReducer(readPreference, topologyDescription, servers.filter(filter)) ) ); if (mode === ReadPreference.SECONDARY_PREFERRED && selectedServers.length === 0) { return servers.filter(primaryFilter); } return selectedServers; }; }