opusdev/vector-similarity-api
1
1import { EJSON, type ObjectId } from '../bson';2import * as WIRE_CONSTANTS from '../cmap/wire_protocol/constants';3import { type MongoError, MongoRuntimeError, MongoStalePrimaryError } from '../error';4import { compareObjectId, shuffle } from '../utils';5import { ServerType, TopologyType } from './common';6import { ServerDescription } from './server_description';7import type { SrvPollingEvent } from './srv_polling';8 9// constants related to compatibility checks10const MIN_SUPPORTED_SERVER_VERSION = WIRE_CONSTANTS.MIN_SUPPORTED_SERVER_VERSION;11const MAX_SUPPORTED_SERVER_VERSION = WIRE_CONSTANTS.MAX_SUPPORTED_SERVER_VERSION;12const MIN_SUPPORTED_WIRE_VERSION = WIRE_CONSTANTS.MIN_SUPPORTED_WIRE_VERSION;13const MAX_SUPPORTED_WIRE_VERSION = WIRE_CONSTANTS.MAX_SUPPORTED_WIRE_VERSION;14 15const MONGOS_OR_UNKNOWN = new Set<ServerType>([ServerType.Mongos, ServerType.Unknown]);16const MONGOS_OR_STANDALONE = new Set<ServerType>([ServerType.Mongos, ServerType.Standalone]);17const NON_PRIMARY_RS_MEMBERS = new Set<ServerType>([18 ServerType.RSSecondary,19 ServerType.RSArbiter,20 ServerType.RSOther21]);22 23/** @public */24export interface TopologyDescriptionOptions {25 heartbeatFrequencyMS?: number;26 localThresholdMS?: number;27}28 29/**30 * Representation of a deployment of servers31 * @public32 */33export class TopologyDescription {34 type: TopologyType;35 setName: string | null;36 maxSetVersion: number | null;37 maxElectionId: ObjectId | null;38 servers: Map<string, ServerDescription>;39 stale: boolean;40 compatible: boolean;41 compatibilityError?: string;42 logicalSessionTimeoutMinutes: number | null;43 heartbeatFrequencyMS: number;44 localThresholdMS: number;45 commonWireVersion: number;46 /**47 * Create a TopologyDescription48 */49 constructor(50 topologyType: TopologyType,51 serverDescriptions: Map<string, ServerDescription> | null = null,52 setName: string | null = null,53 maxSetVersion: number | null = null,54 maxElectionId: ObjectId | null = null,55 commonWireVersion: number | null = null,56 options: TopologyDescriptionOptions | null = null57 ) {58 options = options ?? {};59 60 this.type = topologyType ?? TopologyType.Unknown;61 this.servers = serverDescriptions ?? new Map();62 this.stale = false;63 this.compatible = true;64 this.heartbeatFrequencyMS = options.heartbeatFrequencyMS ?? 0;65 this.localThresholdMS = options.localThresholdMS ?? 15;66 this.setName = setName ?? null;67 this.maxElectionId = maxElectionId ?? null;68 this.maxSetVersion = maxSetVersion ?? null;69 this.commonWireVersion = commonWireVersion ?? 0;70 71 // determine server compatibility72 for (const serverDescription of this.servers.values()) {73 // Load balancer mode is always compatible.74 if (75 serverDescription.type === ServerType.Unknown ||76 serverDescription.type === ServerType.LoadBalancer77 ) {78 continue;79 }80 81 if (serverDescription.minWireVersion > MAX_SUPPORTED_WIRE_VERSION) {82 this.compatible = false;83 this.compatibilityError = `Server at ${serverDescription.address} requires wire version ${serverDescription.minWireVersion}, but this version of the driver only supports up to ${MAX_SUPPORTED_WIRE_VERSION} (MongoDB ${MAX_SUPPORTED_SERVER_VERSION})`;84 }85 86 if (serverDescription.maxWireVersion < MIN_SUPPORTED_WIRE_VERSION) {87 this.compatible = false;88 this.compatibilityError = `Server at ${serverDescription.address} reports wire version ${serverDescription.maxWireVersion}, but this version of the driver requires at least ${MIN_SUPPORTED_WIRE_VERSION} (MongoDB ${MIN_SUPPORTED_SERVER_VERSION}).`;89 break;90 }91 }92 93 // Whenever a client updates the TopologyDescription from a hello response, it MUST set94 // TopologyDescription.logicalSessionTimeoutMinutes to the smallest logicalSessionTimeoutMinutes95 // value among ServerDescriptions of all data-bearing server types. If any have a null96 // logicalSessionTimeoutMinutes, then TopologyDescription.logicalSessionTimeoutMinutes MUST be97 // set to null.98 this.logicalSessionTimeoutMinutes = null;99 for (const [, server] of this.servers) {100 if (server.isReadable) {101 if (server.logicalSessionTimeoutMinutes == null) {102 // If any of the servers have a null logicalSessionsTimeout, then the whole topology does103 this.logicalSessionTimeoutMinutes = null;104 break;105 }106 107 if (this.logicalSessionTimeoutMinutes == null) {108 // First server with a non null logicalSessionsTimeout109 this.logicalSessionTimeoutMinutes = server.logicalSessionTimeoutMinutes;110 continue;111 }112 113 // Always select the smaller of the:114 // current server logicalSessionsTimeout and the topologies logicalSessionsTimeout115 this.logicalSessionTimeoutMinutes = Math.min(116 this.logicalSessionTimeoutMinutes,117 server.logicalSessionTimeoutMinutes118 );119 }120 }121 }122 123 /**124 * Returns a new TopologyDescription based on the SrvPollingEvent125 * @internal126 */127 updateFromSrvPollingEvent(ev: SrvPollingEvent, srvMaxHosts = 0): TopologyDescription {128 /** The SRV addresses defines the set of addresses we should be using */129 const incomingHostnames = ev.hostnames();130 const currentHostnames = new Set(this.servers.keys());131 132 const hostnamesToAdd = new Set<string>(incomingHostnames);133 const hostnamesToRemove = new Set<string>();134 for (const hostname of currentHostnames) {135 // filter hostnamesToAdd (made from incomingHostnames) down to what is *not* present in currentHostnames136 hostnamesToAdd.delete(hostname);137 if (!incomingHostnames.has(hostname)) {138 // If the SRV Records no longer include this hostname139 // we have to stop using it140 hostnamesToRemove.add(hostname);141 }142 }143 144 if (hostnamesToAdd.size === 0 && hostnamesToRemove.size === 0) {145 // No new hosts to add and none to remove146 return this;147 }148 149 const serverDescriptions = new Map(this.servers);150 for (const removedHost of hostnamesToRemove) {151 serverDescriptions.delete(removedHost);152 }153 154 if (hostnamesToAdd.size > 0) {155 if (srvMaxHosts === 0) {156 // Add all!157 for (const hostToAdd of hostnamesToAdd) {158 serverDescriptions.set(hostToAdd, new ServerDescription(hostToAdd));159 }160 } else if (serverDescriptions.size < srvMaxHosts) {161 // Add only the amount needed to get us back to srvMaxHosts162 const selectedHosts = shuffle(hostnamesToAdd, srvMaxHosts - serverDescriptions.size);163 for (const selectedHostToAdd of selectedHosts) {164 serverDescriptions.set(selectedHostToAdd, new ServerDescription(selectedHostToAdd));165 }166 }167 }168 169 return new TopologyDescription(170 this.type,171 serverDescriptions,172 this.setName,173 this.maxSetVersion,174 this.maxElectionId,175 this.commonWireVersion,176 { heartbeatFrequencyMS: this.heartbeatFrequencyMS, localThresholdMS: this.localThresholdMS }177 );178 }179 180 /**181 * Returns a copy of this description updated with a given ServerDescription182 * @internal183 */184 update(serverDescription: ServerDescription): TopologyDescription {185 const address = serverDescription.address;186 187 // potentially mutated values188 let { type: topologyType, setName, maxSetVersion, maxElectionId, commonWireVersion } = this;189 190 const serverType = serverDescription.type;191 const serverDescriptions = new Map(this.servers);192 193 // update common wire version194 if (serverDescription.maxWireVersion !== 0) {195 if (commonWireVersion == null) {196 commonWireVersion = serverDescription.maxWireVersion;197 } else {198 commonWireVersion = Math.min(commonWireVersion, serverDescription.maxWireVersion);199 }200 }201 202 if (203 typeof serverDescription.setName === 'string' &&204 typeof setName === 'string' &&205 serverDescription.setName !== setName206 ) {207 if (topologyType === TopologyType.Single) {208 // "Single" Topology with setName mismatch is direct connection usage, mark unknown do not remove209 serverDescription = new ServerDescription(address);210 } else {211 serverDescriptions.delete(address);212 }213 }214 215 // update the actual server description216 serverDescriptions.set(address, serverDescription);217 218 if (topologyType === TopologyType.Single) {219 // once we are defined as single, that never changes220 return new TopologyDescription(221 TopologyType.Single,222 serverDescriptions,223 setName,224 maxSetVersion,225 maxElectionId,226 commonWireVersion,227 { heartbeatFrequencyMS: this.heartbeatFrequencyMS, localThresholdMS: this.localThresholdMS }228 );229 }230 231 if (topologyType === TopologyType.Unknown) {232 if (serverType === ServerType.Standalone && this.servers.size !== 1) {233 serverDescriptions.delete(address);234 } else {235 topologyType = topologyTypeForServerType(serverType);236 }237 }238 239 if (topologyType === TopologyType.Sharded) {240 if (!MONGOS_OR_UNKNOWN.has(serverType)) {241 serverDescriptions.delete(address);242 }243 }244 245 if (topologyType === TopologyType.ReplicaSetNoPrimary) {246 if (MONGOS_OR_STANDALONE.has(serverType)) {247 serverDescriptions.delete(address);248 }249 250 if (serverType === ServerType.RSPrimary) {251 const result = updateRsFromPrimary(252 serverDescriptions,253 serverDescription,254 setName,255 maxSetVersion,256 maxElectionId257 );258 259 topologyType = result[0];260 setName = result[1];261 maxSetVersion = result[2];262 maxElectionId = result[3];263 } else if (NON_PRIMARY_RS_MEMBERS.has(serverType)) {264 const result = updateRsNoPrimaryFromMember(serverDescriptions, serverDescription, setName);265 topologyType = result[0];266 setName = result[1];267 }268 }269 270 if (topologyType === TopologyType.ReplicaSetWithPrimary) {271 if (MONGOS_OR_STANDALONE.has(serverType)) {272 serverDescriptions.delete(address);273 topologyType = checkHasPrimary(serverDescriptions);274 } else if (serverType === ServerType.RSPrimary) {275 const result = updateRsFromPrimary(276 serverDescriptions,277 serverDescription,278 setName,279 maxSetVersion,280 maxElectionId281 );282 283 topologyType = result[0];284 setName = result[1];285 maxSetVersion = result[2];286 maxElectionId = result[3];287 } else if (NON_PRIMARY_RS_MEMBERS.has(serverType)) {288 topologyType = updateRsWithPrimaryFromMember(289 serverDescriptions,290 serverDescription,291 setName292 );293 } else {294 topologyType = checkHasPrimary(serverDescriptions);295 }296 }297 298 return new TopologyDescription(299 topologyType,300 serverDescriptions,301 setName,302 maxSetVersion,303 maxElectionId,304 commonWireVersion,305 { heartbeatFrequencyMS: this.heartbeatFrequencyMS, localThresholdMS: this.localThresholdMS }306 );307 }308 309 get error(): MongoError | null {310 const descriptionsWithError = Array.from(this.servers.values()).filter(311 (sd: ServerDescription) => sd.error312 );313 314 if (descriptionsWithError.length > 0) {315 return descriptionsWithError[0].error;316 }317 318 return null;319 }320 321 /**322 * Determines if the topology description has any known servers323 */324 get hasKnownServers(): boolean {325 return Array.from(this.servers.values()).some(326 (sd: ServerDescription) => sd.type !== ServerType.Unknown327 );328 }329 330 /**331 * Determines if this topology description has a data-bearing server available.332 */333 get hasDataBearingServers(): boolean {334 return Array.from(this.servers.values()).some((sd: ServerDescription) => sd.isDataBearing);335 }336 337 /**338 * Determines if the topology has a definition for the provided address339 * @internal340 */341 hasServer(address: string): boolean {342 return this.servers.has(address);343 }344 345 /**346 * Returns a JSON-serializable representation of the TopologyDescription. This is primarily347 * intended for use with JSON.stringify().348 *349 * This method will not throw.350 */351 toJSON() {352 return EJSON.serialize(this);353 }354}355 356function topologyTypeForServerType(serverType: ServerType): TopologyType {357 switch (serverType) {358 case ServerType.Standalone:359 return TopologyType.Single;360 case ServerType.Mongos:361 return TopologyType.Sharded;362 case ServerType.RSPrimary:363 return TopologyType.ReplicaSetWithPrimary;364 case ServerType.RSOther:365 case ServerType.RSSecondary:366 return TopologyType.ReplicaSetNoPrimary;367 default:368 return TopologyType.Unknown;369 }370}371 372function updateRsFromPrimary(373 serverDescriptions: Map<string, ServerDescription>,374 serverDescription: ServerDescription,375 setName: string | null = null,376 maxSetVersion: number | null = null,377 maxElectionId: ObjectId | null = null378): [TopologyType, string | null, number | null, ObjectId | null] {379 const setVersionElectionIdMismatch = (380 serverDescription: ServerDescription,381 maxSetVersion: number | null,382 maxElectionId: ObjectId | null383 ) => {384 return (385 `primary marked stale due to electionId/setVersion mismatch:` +386 ` server setVersion: ${serverDescription.setVersion},` +387 ` server electionId: ${serverDescription.electionId},` +388 ` topology setVersion: ${maxSetVersion},` +389 ` topology electionId: ${maxElectionId}`390 );391 };392 setName = setName || serverDescription.setName;393 if (setName !== serverDescription.setName) {394 serverDescriptions.delete(serverDescription.address);395 return [checkHasPrimary(serverDescriptions), setName, maxSetVersion, maxElectionId];396 }397 398 if (serverDescription.maxWireVersion >= 17) {399 const electionIdComparison = compareObjectId(maxElectionId, serverDescription.electionId);400 const maxElectionIdIsEqual = electionIdComparison === 0;401 const maxElectionIdIsLess = electionIdComparison === -1;402 const maxSetVersionIsLessOrEqual =403 (maxSetVersion ?? -1) <= (serverDescription.setVersion ?? -1);404 405 if (maxElectionIdIsLess || (maxElectionIdIsEqual && maxSetVersionIsLessOrEqual)) {406 // The reported electionId was greater407 // or the electionId was equal and reported setVersion was greater408 // Always update both values, they are a tuple409 maxElectionId = serverDescription.electionId;410 maxSetVersion = serverDescription.setVersion;411 } else {412 // Stale primary413 // replace serverDescription with a default ServerDescription of type "Unknown"414 serverDescriptions.set(415 serverDescription.address,416 new ServerDescription(serverDescription.address, undefined, {417 error: new MongoStalePrimaryError(418 setVersionElectionIdMismatch(serverDescription, maxSetVersion, maxElectionId)419 )420 })421 );422 423 return [checkHasPrimary(serverDescriptions), setName, maxSetVersion, maxElectionId];424 }425 } else {426 const electionId = serverDescription.electionId ? serverDescription.electionId : null;427 if (serverDescription.setVersion && electionId) {428 if (maxSetVersion && maxElectionId) {429 if (430 maxSetVersion > serverDescription.setVersion ||431 compareObjectId(maxElectionId, electionId) > 0432 ) {433 // this primary is stale, we must remove it434 serverDescriptions.set(435 serverDescription.address,436 new ServerDescription(serverDescription.address, undefined, {437 error: new MongoStalePrimaryError(438 setVersionElectionIdMismatch(serverDescription, maxSetVersion, maxElectionId)439 )440 })441 );442 443 return [checkHasPrimary(serverDescriptions), setName, maxSetVersion, maxElectionId];444 }445 }446 447 maxElectionId = serverDescription.electionId;448 }449 450 if (451 serverDescription.setVersion != null &&452 (maxSetVersion == null || serverDescription.setVersion > maxSetVersion)453 ) {454 maxSetVersion = serverDescription.setVersion;455 }456 }457 458 // We've heard from the primary. Is it the same primary as before?459 for (const [address, server] of serverDescriptions) {460 if (server.type === ServerType.RSPrimary && server.address !== serverDescription.address) {461 // Reset old primary's type to Unknown.462 serverDescriptions.set(463 address,464 new ServerDescription(server.address, undefined, {465 error: new MongoStalePrimaryError(466 'primary marked stale due to discovery of newer primary'467 )468 })469 );470 471 // There can only be one primary472 break;473 }474 }475 476 // Discover new hosts from this primary's response.477 serverDescription.allHosts.forEach((address: string) => {478 if (!serverDescriptions.has(address)) {479 serverDescriptions.set(address, new ServerDescription(address));480 }481 });482 483 // Remove hosts not in the response.484 const currentAddresses = Array.from(serverDescriptions.keys());485 const responseAddresses = serverDescription.allHosts;486 currentAddresses487 .filter((addr: string) => responseAddresses.indexOf(addr) === -1)488 .forEach((address: string) => {489 serverDescriptions.delete(address);490 });491 492 return [checkHasPrimary(serverDescriptions), setName, maxSetVersion, maxElectionId];493}494 495function updateRsWithPrimaryFromMember(496 serverDescriptions: Map<string, ServerDescription>,497 serverDescription: ServerDescription,498 setName: string | null = null499): TopologyType {500 if (setName == null) {501 // TODO(NODE-3483): should be an appropriate runtime error502 throw new MongoRuntimeError('Argument "setName" is required if connected to a replica set');503 }504 505 if (506 setName !== serverDescription.setName ||507 (serverDescription.me && serverDescription.address !== serverDescription.me)508 ) {509 serverDescriptions.delete(serverDescription.address);510 }511 512 return checkHasPrimary(serverDescriptions);513}514 515function updateRsNoPrimaryFromMember(516 serverDescriptions: Map<string, ServerDescription>,517 serverDescription: ServerDescription,518 setName: string | null = null519): [TopologyType, string | null] {520 const topologyType = TopologyType.ReplicaSetNoPrimary;521 setName = setName ?? serverDescription.setName;522 if (setName !== serverDescription.setName) {523 serverDescriptions.delete(serverDescription.address);524 return [topologyType, setName];525 }526 527 serverDescription.allHosts.forEach((address: string) => {528 if (!serverDescriptions.has(address)) {529 serverDescriptions.set(address, new ServerDescription(address));530 }531 });532 533 if (serverDescription.me && serverDescription.address !== serverDescription.me) {534 serverDescriptions.delete(serverDescription.address);535 }536 537 return [topologyType, setName];538}539 540function checkHasPrimary(serverDescriptions: Map<string, ServerDescription>): TopologyType {541 for (const serverDescription of serverDescriptions.values()) {542 if (serverDescription.type === ServerType.RSPrimary) {543 return TopologyType.ReplicaSetWithPrimary;544 }545 }546 547 return TopologyType.ReplicaSetNoPrimary;548}549