-
Notifications
You must be signed in to change notification settings - Fork 585
Expand file tree
/
Copy pathchannelCollection.ts
More file actions
1938 lines (1782 loc) · 66 KB
/
Copy pathchannelCollection.ts
File metadata and controls
1938 lines (1782 loc) · 66 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*!
* Copyright (c) Microsoft Corporation and contributors. All rights reserved.
* Licensed under the MIT License.
*/
import { AttachState } from "@fluidframework/container-definitions";
import type {
FluidObject,
IDisposable,
IRequest,
IResponse,
ITelemetryBaseLogger,
} from "@fluidframework/core-interfaces";
import type {
IFluidHandleInternal,
ISignalEnvelope,
} from "@fluidframework/core-interfaces/internal";
import { assert, Lazy, LazyPromise } from "@fluidframework/core-utils/internal";
import { FluidObjectHandle } from "@fluidframework/datastore/internal";
import type {
ISnapshot,
ISnapshotTree,
ISequencedDocumentMessage,
} from "@fluidframework/driver-definitions/internal";
import {
buildSnapshotTree,
getSnapshotTree,
isInstanceOfISnapshot,
} from "@fluidframework/driver-utils/internal";
import type {
AliasResult,
ContainerExtensionProvider,
FluidDataStoreMessage,
IAttachMessage,
IEnvelope,
IFluidDataStoreChannel,
IFluidDataStoreContext,
IFluidDataStoreContextDetached,
IFluidDataStoreFactory,
IFluidDataStoreRegistry,
IFluidParentContext,
IGarbageCollectionData,
IInboundSignalMessage,
InboundAttachMessage,
IRuntimeMessageCollection,
IRuntimeMessagesContent,
ISummarizeResult,
ISummaryTreeWithStats,
ITelemetryContext,
MinimumVersionForCollab,
NamedFluidDataStoreRegistryEntries,
} from "@fluidframework/runtime-definitions/internal";
import {
CreateSummarizerNodeSource,
channelsTreeName,
gcDataBlobKey,
} from "@fluidframework/runtime-definitions/internal";
import {
GCDataBuilder,
RequestParser,
RuntimeHeaders,
SummaryTreeBuilder,
addBlobToSummary,
convertSnapshotTreeToSummaryTree,
convertSummaryTreeToITree,
create404Response,
createResponseError,
encodeCompactIdToString,
forEachContiguousBunch,
isSerializedHandle,
processAttachMessageGCData,
responseToException,
unpackChildNodesUsedRoutes,
} from "@fluidframework/runtime-utils/internal";
import {
DataCorruptionError,
DataProcessingError,
LoggingError,
type MonitoringContext,
createChildLogger,
createChildMonitoringContext,
extractSafePropertiesFromMessage,
tagCodeArtifacts,
type ITelemetryPropertiesExt,
} from "@fluidframework/telemetry-utils/internal";
import { v4 as uuid } from "uuid";
import {
DeletedResponseHeaderKey,
type RuntimeHeaderData,
defaultRuntimeHeaderData,
} from "./containerRuntime.js";
import {
type IDataStoreAliasMessage,
channelToDataStore,
isDataStoreAliasMessage,
} from "./dataStore.js";
import {
FluidDataStoreContext,
type IFluidDataStoreContextPrivate,
type ILocalDetachedFluidDataStoreContextProps,
LocalDetachedFluidDataStoreContext,
LocalFluidDataStoreContext,
RemoteFluidDataStoreContext,
createAttributesBlob,
} from "./dataStoreContext.js";
import { DataStoreContexts } from "./dataStoreContexts.js";
import { FluidDataStoreRegistry } from "./dataStoreRegistry.js";
import { GCNodeType, type IGCNodeUpdatedProps, urlToGCNodePath } from "./gc/index.js";
import type {
ContainerRuntimeAliasMessage,
ContainerRuntimeDataStoreOpMessage,
OutboundContainerRuntimeAttachMessage,
} from "./messageTypes.js";
import { ContainerMessageType, type LocalContainerRuntimeMessage } from "./messageTypes.js";
import { StorageServiceWithAttachBlobs } from "./storageServiceWithAttachBlobs.js";
import {
type IContainerRuntimeMetadata,
nonDataStorePaths,
rootHasIsolatedChannels,
} from "./summary/index.js";
/**
* True if a tombstoned object should be returned without erroring
* @legacy @beta
*/
export const AllowTombstoneRequestHeaderKey = "allowTombstone"; // Belongs in the enum above, but avoiding the breaking change
type PendingAliasResolve = (success: boolean) => void;
/**
* Envelope for signals not intended for the container.
*
* @privateRemarks
* `clientBroadcastSignalSequenceNumber` might be added to the envelope by the container runtime.
* But it should not be provided to start with.
*
* Equivalent to `Required<Omit<ISignalEnvelope, "clientBroadcastSignalSequenceNumber">>`.
*/
export type AddressedUnsequencedSignalEnvelope = IEnvelope<ISignalEnvelope["contents"]>;
/**
* This version of the interface is private to this the package. It should never be exported under any tag.
* It is used to manage interactions within the container-runtime package. If something is needed
* cross package, it is likely it is also being used cross layer (ContainerRuntime * DataStoreRuntime).
* If that is the case, the change likely needs to be staged directly on IFluidParentContext. Changes
* being staged on IFluidParentContext can be added here as well, likely with optionality removed,
* to ease interactions within this package.
*/
export interface IFluidParentContextPrivate
extends IFluidParentContext,
ContainerExtensionProvider {
readonly isReadOnly: () => boolean;
readonly minVersionForCollab: MinimumVersionForCollab;
}
/**
* Kin of {@link @fluidframework/runtime-definitions#IFluidParentContext} with alternate
* `submitMessage` and `submitSignal` methods that are typed specifically for the
* root context (aka {@link ContainerRuntime} provided context).
*
* @privateRemarks
* These replacements might be able to get cleaned up if the future suggestions
* found in {@link @fluidframework/runtime-definitions#FluidDataStoreMessage}
* `@privateRemarks` section are implemented.
*/
export interface IFluidRootParentContextPrivate
extends Omit<IFluidParentContextPrivate, "submitMessage" | "submitSignal"> {
/**
* Submits the message to be sent to other clients.
* @param containerRuntimeMessage - The message.
* @param localOpMetadata - The local metadata associated with the message.
* This is kept locally and not sent to the server. This will be sent back
* when this message is received back from the server. This is also sent if
* we are asked to resubmit the message.
*/
readonly submitMessage: (
containerRuntimeMessage:
| ContainerRuntimeDataStoreOpMessage
| OutboundContainerRuntimeAttachMessage
| ContainerRuntimeAliasMessage,
localOpMetadata: unknown,
) => void;
/**
* Submits the signal to be sent to other clients.
* @param envelope - {@link IEnvelope} containing the signal address and contents.
* @param targetClientId - When specified, the signal is only sent to the provided client id.
*/
readonly submitSignal: (
envelope: AddressedUnsequencedSignalEnvelope,
targetClientId?: string,
) => void;
}
type SubmitKeys = "submitMessage" | "submitSignal";
/**
* Creates a shallow wrapper of {@link IFluidParentContextPrivate} or
* {@link IFluidRootParentContextPrivate} with `submitMessage` and `submitSignal`
* methods replaced with the provided overrides.
*/
export function formParentContext<
T extends IFluidParentContextPrivate | IFluidRootParentContextPrivate,
>(
context: Omit<IFluidParentContextPrivate & IFluidRootParentContextPrivate, SubmitKeys>,
overrides: Pick<T, SubmitKeys>,
): Omit<IFluidParentContextPrivate & IFluidRootParentContextPrivate, SubmitKeys> &
Pick<T, SubmitKeys> {
return {
get IFluidDataStoreRegistry() {
return context.IFluidDataStoreRegistry;
},
IFluidHandleContext: context.IFluidHandleContext,
options: context.options,
get clientId() {
return context.clientId;
},
get connected() {
return context.connected;
},
deltaManager: context.deltaManager,
storage: context.storage,
baseLogger: context.baseLogger,
get clientDetails() {
return context.clientDetails;
},
get idCompressor() {
return context.idCompressor;
},
loadingGroupId: context.loadingGroupId,
get attachState() {
return context.attachState;
},
isReadOnly: () => context.isReadOnly(),
containerRuntime: context.containerRuntime,
scope: context.scope,
gcThrowOnTombstoneUsage: context.gcThrowOnTombstoneUsage,
gcTombstoneEnforcementAllowed: context.gcTombstoneEnforcementAllowed,
getAbsoluteUrl: async (...args) => {
return context.getAbsoluteUrl(...args);
},
getQuorum: (...args) => {
return context.getQuorum(...args);
},
getAudience: (...args) => {
return context.getAudience(...args);
},
submitMessage: overrides.submitMessage.bind(overrides),
submitSignal: overrides.submitSignal,
makeLocallyVisible: (...args) => {
return context.makeLocallyVisible(...args);
},
uploadBlob: async (...args) => {
return context.uploadBlob(...args);
},
addedGCOutboundRoute: (...args) => {
return context.addedGCOutboundRoute(...args);
},
getCreateChildSummarizerNodeFn: (...args) => {
return context.getCreateChildSummarizerNodeFn?.(...args);
},
deleteChildSummarizerNode: (...args) => {
return context.deleteChildSummarizerNode(...args);
},
setChannelDirty: (address: string) => {
return context.setChannelDirty(address);
},
minVersionForCollab: context.minVersionForCollab,
getExtension: context.getExtension.bind(context),
};
}
/**
* Creates a wrapper of a {@link IFluidRootParentContextPrivate} to be provided to the inner datastore channels.
* The wrapper will have the submit methods overwritten with the appropriate id as the destination address.
*
* @param id - the id of the channel
* @param parentContext - the {@link IFluidRootParentContextPrivate} to wrap
* @returns A wrapped {@link IFluidParentContext}
*/
function wrapContextForInnerChannel(
id: string,
parentContext: IFluidRootParentContextPrivate,
): IFluidParentContextPrivate {
const context = formParentContext<IFluidParentContextPrivate>(parentContext, {
submitMessage: (type: string, content: unknown, localOpMetadata: unknown) => {
const fluidDataStoreContent: FluidDataStoreMessage = {
content,
type,
};
const envelope = {
address: id,
contents: fluidDataStoreContent,
};
parentContext.submitMessage(
{ type: ContainerMessageType.FluidDataStoreOp, contents: envelope },
localOpMetadata,
);
},
submitSignal: (type: string, content: unknown, targetClientId?: string) => {
parentContext.submitSignal({ address: id, contents: { type, content } }, targetClientId);
},
});
return context;
}
/**
* Returns the type of the given local data store from its package path.
*/
export function getLocalDataStoreType(localDataStore: LocalFluidDataStoreContext): string {
return localDataStore.packagePath[localDataStore.packagePath.length - 1];
}
/**
* This class encapsulates data store handling. Currently it is only used by the container runtime,
* but eventually could be hosted on any channel once we formalize the channel api boundary.
* @internal
*/
export class ChannelCollection
implements Omit<IFluidDataStoreChannel, "entryPoint" | "reSubmit" | "rollback">, IDisposable
{
// Stores tracked by the Domain
private readonly pendingAttach = new Map<string, IAttachMessage>();
// 0.24 back-compat attachingBeforeSummary
public readonly attachOpFiredForDataStore = new Set<string>();
protected readonly mc: MonitoringContext;
// eslint-disable-next-line unicorn/consistent-function-scoping -- Property is defined once; no need to extract inner lambda
private readonly disposeOnce = new Lazy<void>(() => this.contexts.dispose());
public readonly containerLoadStats: {
// number of dataStores during loadContainer
readonly containerLoadDataStoreCount: number;
// number of unreferenced dataStores during loadContainer
readonly referencedDataStoreCount: number;
};
private readonly pendingAliasMap: Map<string, Promise<AliasResult>> = new Map<
string,
Promise<AliasResult>
>();
protected readonly contexts: DataStoreContexts;
private readonly aliasedDataStores: Set<string>;
constructor(
protected readonly baseSnapshot: ISnapshotTree | ISnapshot | undefined,
public readonly parentContext: IFluidRootParentContextPrivate,
baseLogger: ITelemetryBaseLogger,
private readonly gcNodeUpdated: (props: IGCNodeUpdatedProps) => void,
private readonly isDataStoreDeleted: (nodePath: string) => boolean,
private readonly aliasMap: Map<string, string>,
) {
this.mc = createChildMonitoringContext({ logger: baseLogger });
this.contexts = new DataStoreContexts(baseLogger);
this.aliasedDataStores = new Set(aliasMap.values());
// Extract stores stored inside the snapshot
const fluidDataStores = new Map<string, ISnapshotTree>();
if (baseSnapshot) {
const baseSnapshotTree = getSnapshotTree(baseSnapshot);
for (const [key, value] of Object.entries(baseSnapshotTree.trees)) {
fluidDataStores.set(key, value);
}
}
let unreferencedDataStoreCount = 0;
// Create a context for each of them
for (const [key, value] of fluidDataStores) {
let dataStoreContext: FluidDataStoreContext;
// counting number of unreferenced data stores
if (value.unreferenced) {
unreferencedDataStoreCount++;
}
// If we have a detached container, then create local data store contexts.
if (this.parentContext.attachState === AttachState.Detached) {
if (typeof value !== "object") {
throw new LoggingError("Snapshot should be there to load from!!");
}
const snapshotTree = value;
dataStoreContext = new LocalFluidDataStoreContext({
id: key,
pkg: undefined,
parentContext: this.wrapContextForInnerChannel(key),
storage: this.parentContext.storage,
scope: this.parentContext.scope,
createSummarizerNodeFn: this.parentContext.getCreateChildSummarizerNodeFn(key, {
type: CreateSummarizerNodeSource.FromSummary,
}),
makeLocallyVisibleFn: () => this.makeDataStoreLocallyVisible(key),
snapshotTree,
});
} else {
let snapshotForRemoteFluidDatastoreContext: ISnapshot | ISnapshotTree = value;
if (isInstanceOfISnapshot(baseSnapshot)) {
snapshotForRemoteFluidDatastoreContext = {
...baseSnapshot,
snapshotTree: value,
};
}
dataStoreContext = new RemoteFluidDataStoreContext({
id: key,
snapshot: snapshotForRemoteFluidDatastoreContext,
parentContext: this.wrapContextForInnerChannel(key),
storage: this.parentContext.storage,
scope: this.parentContext.scope,
createSummarizerNodeFn: this.parentContext.getCreateChildSummarizerNodeFn(key, {
type: CreateSummarizerNodeSource.FromSummary,
}),
loadingGroupId: value.groupId,
});
}
this.contexts.addBoundOrRemoted(dataStoreContext);
}
this.containerLoadStats = {
containerLoadDataStoreCount: fluidDataStores.size,
referencedDataStoreCount: fluidDataStores.size - unreferencedDataStoreCount,
};
}
public get aliases(): ReadonlyMap<string, string> {
return this.aliasMap;
}
public get pendingAliases(): Map<string, Promise<AliasResult>> {
return this.pendingAliasMap;
}
public async waitIfPendingAlias(maybeAlias: string): Promise<AliasResult> {
const pendingAliasPromise = this.pendingAliases.get(maybeAlias);
return pendingAliasPromise ?? "Success";
}
/**
* For sampling. Only log once per container
*/
private shouldSendAttachLog = true;
protected wrapContextForInnerChannel(id: string): IFluidParentContextPrivate {
return wrapContextForInnerChannel(id, this.parentContext);
}
/**
* IFluidDataStoreChannel.makeVisibleAndAttachGraph implementation
* Not clear when it would be called and what it should do.
* Currently this API is called by context only for root data stores.
*/
public makeVisibleAndAttachGraph(): void {
this.parentContext.makeLocallyVisible();
}
private processAttachMessages(messageCollection: IRuntimeMessageCollection): void {
const { envelope, messagesContent, local } = messageCollection;
for (const { contents } of messagesContent) {
const attachMessage = contents as InboundAttachMessage;
// We need to process the GC Data for both local and remote attach messages
const foundGCData = processAttachMessageGCData(
attachMessage.snapshot ?? undefined,
(nodeId, toPath) => {
// nodeId is the relative path under the node being attached. Always starts with "/", but no trailing "/" after an id
const fromPath = `/${attachMessage.id}${nodeId === "/" ? "" : nodeId}`;
this.parentContext.addedGCOutboundRoute(fromPath, toPath, envelope.timestamp);
},
);
// Only log once per container to avoid noise/cost.
// Allows longitudinal tracking of various state (e.g. foundGCData), and some sampled details
if (this.shouldSendAttachLog) {
this.shouldSendAttachLog = false;
this.mc.logger.sendTelemetryEvent({
eventName: "dataStoreAttachMessage_sampled",
...tagCodeArtifacts({ id: attachMessage.id, pkg: attachMessage.type }),
details: {
local,
snapshot: !!attachMessage.snapshot,
foundGCData,
},
...extractSafePropertiesFromMessage(envelope),
});
}
// The local object has already been attached
if (local) {
assert(
this.pendingAttach.has(attachMessage.id),
0x15e /* "Local object does not have matching attach message id" */,
);
this.contexts.get(attachMessage.id)?.setAttachState(AttachState.Attached);
this.pendingAttach.delete(attachMessage.id);
continue;
}
// Check for collision with local (not yet live / known to other clients) DataStore
// This is not a DataCorruption case if we crash the container before the DataStore becomes visible to others (it's a DataProcessingError instead)
//
// POSSIBLE CAUSES:
// - Something with ID creation, e.g. a bug in shortID logic, or somehow a generated ID matches an existing alias.
// - An invalid operation by the application or service where an existing container is returned to a new container attach call,
// resulting in duplicate accounting for objects that were supposed to be local-only. e.g. if the application patches in custom
// logic not supported by Fluid's API.
if (this.contexts.getUnbound(attachMessage.id) !== undefined) {
const error = DataProcessingError.create(
"Local DataStore matches remote DataStore id",
"DataStoreAttach",
envelope,
{ ...tagCodeArtifacts({ dataStoreId: attachMessage.id }) },
);
throw error;
}
// Check for collision with already processed (attaching/attached or aliased) DataStore
// This is presumed to indicate a corrupted op stream, where we'd expect all future sessions to fail here too.
//
// POSSIBLE CAUSES:
// - A bug in the service or driver that results in ops being duplicated
// - Similar to above, an existing container being returned to a new container attach call,
// where the DataStore in question was already made locally visible before container attach.
// (Perhaps future sessions would not fail in this case, but it's hypothetical and hard to differentiate)
if (this.alreadyProcessed(attachMessage.id)) {
const error = new DataCorruptionError(
// pre-0.58 error message: duplicateDataStoreCreatedWithExistingId
"Duplicate DataStore created with existing id",
{
...extractSafePropertiesFromMessage(envelope),
...tagCodeArtifacts({ dataStoreId: attachMessage.id }),
},
);
throw error;
}
const flatAttachBlobs = new Map<string, ArrayBufferLike>();
let snapshot: ISnapshotTree | ISnapshot | undefined;
if (attachMessage.snapshot) {
snapshot = buildSnapshotTree(attachMessage.snapshot.entries, flatAttachBlobs);
if (isInstanceOfISnapshot(this.baseSnapshot)) {
snapshot = { ...this.baseSnapshot, snapshotTree: snapshot };
}
}
// Include the type of attach message which is the pkg of the store to be
// used by RemoteFluidDataStoreContext in case it is not in the snapshot.
const pkg = [attachMessage.type];
const remoteFluidDataStoreContext = new RemoteFluidDataStoreContext({
id: attachMessage.id,
snapshot,
parentContext: this.wrapContextForInnerChannel(attachMessage.id),
storage: new StorageServiceWithAttachBlobs(
this.parentContext.storage,
flatAttachBlobs,
),
scope: this.parentContext.scope,
loadingGroupId: attachMessage.snapshot?.groupId,
createSummarizerNodeFn: this.parentContext.getCreateChildSummarizerNodeFn(
attachMessage.id,
{
type: CreateSummarizerNodeSource.FromAttach,
sequenceNumber: envelope.sequenceNumber,
snapshot: attachMessage.snapshot ?? {
entries: [createAttributesBlob(pkg, true /* isRootDataStore */)],
},
},
),
pkg,
});
this.contexts.addBoundOrRemoted(remoteFluidDataStoreContext);
}
}
private processAliasMessages(messageCollection: IRuntimeMessageCollection): void {
const { envelope, messagesContent, local } = messageCollection;
for (const { contents, localOpMetadata } of messagesContent) {
const aliasMessage = contents as IDataStoreAliasMessage;
if (!isDataStoreAliasMessage(aliasMessage)) {
throw new DataCorruptionError("malformedDataStoreAliasMessage", {
...extractSafePropertiesFromMessage(envelope),
});
}
const resolve = localOpMetadata as PendingAliasResolve;
const aliasResult = this.processAliasMessageCore(
aliasMessage.internalId,
aliasMessage.alias,
envelope.timestamp,
);
if (local) {
resolve(aliasResult);
}
}
}
public processAliasMessageCore(
internalId: string,
alias: string,
messageTimestampMs?: number,
): boolean {
if (this.alreadyProcessed(alias)) {
return false;
}
const context = this.contexts.get(internalId);
// If the data store has been deleted, log an error and ignore this message. This helps prevent document
// corruption in case a deleted data store accidentally submitted a signal.
if (this.checkAndLogIfDeleted(internalId, context, "Changed", "processAliasMessageCore")) {
return false;
}
if (context === undefined) {
this.mc.logger.sendErrorEvent({
eventName: "AliasFluidDataStoreNotFound",
fluidDataStoreId: internalId,
});
return false;
}
// If message timestamp doesn't exist, this is called in a detached container. Don't notify GC in that case
// because it doesn't run in detached container and doesn't need to know about this route.
if (messageTimestampMs !== undefined) {
this.parentContext.addedGCOutboundRoute("/", `/${internalId}`, messageTimestampMs);
}
this.aliasMap.set(alias, context.id);
this.aliasedDataStores.add(context.id);
context.setInMemoryRoot();
return true;
}
private alreadyProcessed(id: string): boolean {
return this.aliasMap.get(id) !== undefined || this.contexts.get(id) !== undefined;
}
/**
* Package up the context's attach summary etc into an IAttachMessage
*/
private generateAttachMessage(localContext: LocalFluidDataStoreContext): IAttachMessage {
// Get the attach summary.
const attachSummary = localContext.getAttachSummary();
// Get the GC data and add it to the attach summary.
const attachGCData = localContext.getAttachGCData();
addBlobToSummary(attachSummary, gcDataBlobKey, JSON.stringify(attachGCData));
// Attach message needs the summary in ITree format. Convert the ISummaryTree into an ITree.
const snapshot = convertSummaryTreeToITree(attachSummary.summary);
return {
id: localContext.id,
snapshot,
type: getLocalDataStoreType(localContext),
} satisfies IAttachMessage;
}
/**
* Make the data store locally visible in the container graph by moving the data store context from unbound to
* bound list and submitting the attach message. This data store can now be reached from the root.
* @param id - The id of the data store context to make visible.
*/
private makeDataStoreLocallyVisible(id: string): void {
const localContext = this.contexts.getUnbound(id);
assert(!!localContext, 0x15f /* "Could not find unbound context to bind" */);
/**
* If the container is not detached, it is globally visible to all clients. This data store should also be
* globally visible. Move it to attaching state and send an "attach" op for it.
* If the container is detached, this data store will be part of the summary that makes the container attached.
*/
if (this.parentContext.attachState !== AttachState.Detached) {
this.submitAttachChannelOp(localContext);
localContext.setAttachState(AttachState.Attaching);
}
this.contexts.bind(id);
}
protected submitAttachChannelOp(localContext: LocalFluidDataStoreContext): void {
const message = this.generateAttachMessage(localContext);
this.pendingAttach.set(localContext.id, message);
this.parentContext.submitMessage(
{ type: ContainerMessageType.Attach, contents: message },
undefined,
);
this.attachOpFiredForDataStore.add(localContext.id);
}
/**
* Generate compact internal DataStore ID.
*
* A note about namespace and name collisions:
* This code assumes that that's the only way to generate internal IDs, and that it's Ok for this namespace to overlap with
* user-provided alias names namespace.
* There are two scenarios where it could cause trouble:
* 1) Old files, where (already removed) CreateRoot*DataStore*() API was used, and thus internal name of data store
* was provided by user. Such files may experience name collision with future data stores that receive a name generated
* by this function.
* 2) Much less likely, but if it happen that internal ID (generated by this function) is exactly the same as alias name
* that user might use in the future, them ContainerRuntime.getAliasedDataStoreEntryPoint() or
* ContainerRuntime.getDataStoreFromRequest() could return a data store with internalID matching user request, even though
* user expected some other data store (that would receive alias later).
* Please note that above mentioned functions have the implementation they have (allowing #2) due to #1.
*/
protected createDataStoreId(): string {
/**
* Return uuid if short-ids are explicitly disabled via feature flags.
*/
if (this.mc.config.getBoolean("Fluid.Runtime.DisableShortIds") === true) {
return uuid();
} else {
// We use three non-overlapping namespaces:
// - detached state: even numbers
// - attached state: odd numbers
// - uuids
// In first two cases we will encode result as strings in more compact form.
if (this.parentContext.attachState === AttachState.Detached) {
// container is detached, only one client observes content, no way to hit collisions with other clients.
return encodeCompactIdToString(2 * this.contexts.size);
}
const id = this.parentContext.containerRuntime.generateDocumentUniqueId();
if (typeof id === "number") {
return encodeCompactIdToString(2 * id + 1);
}
return id;
}
}
public createDetachedDataStore(
pkg: readonly string[],
loadingGroupId?: string,
): IFluidDataStoreContextDetached {
return this.createContext(
this.createDataStoreId(),
pkg,
LocalDetachedFluidDataStoreContext,
loadingGroupId,
);
}
public createDataStoreContext(
pkg: readonly string[],
loadingGroupId?: string,
): IFluidDataStoreContextPrivate {
return this.createContext(
this.createDataStoreId(),
pkg,
LocalFluidDataStoreContext,
loadingGroupId,
);
}
protected createContext<T extends LocalFluidDataStoreContext>(
id: string,
pkg: readonly string[],
contextCtor: new (props: ILocalDetachedFluidDataStoreContextProps) => T,
loadingGroupId?: string,
): T {
assert(loadingGroupId !== "", 0x974 /* loadingGroupId should not be the empty string */);
const context = new contextCtor({
id,
pkg,
parentContext: this.wrapContextForInnerChannel(id),
storage: this.parentContext.storage,
scope: this.parentContext.scope,
createSummarizerNodeFn: this.parentContext.getCreateChildSummarizerNodeFn(id, {
type: CreateSummarizerNodeSource.Local,
}),
makeLocallyVisibleFn: () => this.makeDataStoreLocallyVisible(id),
snapshotTree: undefined,
loadingGroupId,
channelToDataStoreFn: (channel: IFluidDataStoreChannel) =>
channelToDataStore(
channel,
id,
this,
createChildLogger({ logger: this.parentContext.baseLogger }),
),
});
this.contexts.addUnbound(context);
return context;
}
public get disposed(): boolean {
return this.disposeOnce.evaluated;
}
public dispose(): void {
return this.disposeOnce.value;
}
public readonly reSubmitContainerMessage = (
message:
| ContainerRuntimeDataStoreOpMessage
| OutboundContainerRuntimeAttachMessage
| ContainerRuntimeAliasMessage,
localOpMetadata: unknown,
squash: boolean,
): void => {
switch (message.type) {
case ContainerMessageType.Attach:
case ContainerMessageType.Alias: {
this.parentContext.submitMessage(message, localOpMetadata);
return;
}
case ContainerMessageType.FluidDataStoreOp: {
return this.resubmitDataStoreOp(message.contents, localOpMetadata, squash);
}
default: {
assert(false, 0x907 /* unknown op type */);
}
}
};
protected readonly resubmitDataStoreOp = (
envelope: IEnvelope<FluidDataStoreMessage>,
localOpMetadata: unknown,
squash: boolean,
): void => {
const context = this.contexts.get(envelope.address);
// If the data store has been deleted, log an error and throw an error. If there are local changes for a
// deleted data store, it can otherwise lead to inconsistent state when compared to other clients.
if (
this.checkAndLogIfDeleted(envelope.address, context, "Changed", "resubmitDataStoreOp")
) {
throw new DataCorruptionError("Context is deleted!", {
callSite: "resubmitDataStoreOp",
...tagCodeArtifacts({ id: envelope.address }),
});
}
assert(!!context, 0x160 /* "There should be a store context for the op" */);
context.reSubmit(envelope.contents, localOpMetadata, squash);
};
/**
* Resubmit a contiguous run of FluidDataStoreOp entries. Entries are bunched by
* `(address, FluidDataStoreMessage.type)` and forwarded to each data store context in a single
* {@link FluidDataStoreContext.reSubmitMessages} call per bunch — mirroring the inbound
* {@link ChannelCollection.processChannelMessages} bunching pattern.
*/
public readonly reSubmitContainerMessages = (
entries: readonly {
envelope: IEnvelope<FluidDataStoreMessage>;
localOpMetadata: unknown;
}[],
squash: boolean,
): void => {
forEachContiguousBunch(
entries,
(e) => ({ address: e.envelope.address, type: e.envelope.contents.type }),
(e) => ({
contents: e.envelope.contents.content,
localOpMetadata: e.localOpMetadata,
}),
(key, messages) => {
const context = this.contexts.get(key.address);
if (
this.checkAndLogIfDeleted(
key.address,
context,
"Changed",
"reSubmitContainerMessages",
)
) {
throw new DataCorruptionError("Context is deleted!", {
callSite: "reSubmitContainerMessages",
...tagCodeArtifacts({ id: key.address }),
});
}
assert(!!context, "There should be a store context for the op");
context.reSubmitMessages(key.type, { squash, messages });
},
(a, b) => a.address === b.address && a.type === b.type,
);
};
public readonly rollbackDataStoreOp = (
envelope: IEnvelope<FluidDataStoreMessage>,
localOpMetadata: unknown,
): void => {
const context = this.contexts.get(envelope.address);
// If the data store has been deleted, log an error and throw an error. If there are local changes for a
// deleted data store, it can otherwise lead to inconsistent state when compared to other clients.
if (
this.checkAndLogIfDeleted(envelope.address, context, "Changed", "rollbackDataStoreOp")
) {
throw new DataCorruptionError("Context is deleted!", {
callSite: "rollbackDataStoreOp",
...tagCodeArtifacts({ id: envelope.address }),
});
}
assert(!!context, 0x2e8 /* "There should be a store context for the op" */);
context.rollback(envelope.contents, localOpMetadata);
};
public async applyStashedOp(content: unknown): Promise<unknown> {
const opContents = content as LocalContainerRuntimeMessage;
switch (opContents.type) {
case ContainerMessageType.Attach: {
return this.applyStashedAttachOp(opContents.contents);
}
case ContainerMessageType.Alias: {
return;
}
case ContainerMessageType.FluidDataStoreOp: {
return this.applyStashedChannelChannelOp(opContents.contents);
}
default: {
assert(false, 0x908 /* unknon type of op */);
}
}
}
protected async applyStashedChannelChannelOp(envelope: IEnvelope): Promise<unknown> {
const context = this.contexts.get(envelope.address);
// If the data store has been deleted, log an error and ignore this message. This helps prevent document
// corruption in case the data store that stashed the op is deleted.
if (this.checkAndLogIfDeleted(envelope.address, context, "Changed", "applyStashedOp")) {
return undefined;
}
assert(!!context, 0x161 /* "There should be a store context for the op" */);
return context.applyStashedOp(envelope.contents);
}
private async applyStashedAttachOp(message: IAttachMessage): Promise<void> {
const { id, snapshot } = message;
// build the snapshot from the summary in the attach message
const flatAttachBlobs = new Map<string, ArrayBufferLike>();
const snapshotTree = buildSnapshotTree(snapshot.entries, flatAttachBlobs);
const storage = new StorageServiceWithAttachBlobs(
this.parentContext.storage,
flatAttachBlobs,
);
// create a local datastore context for the data store context,
// which this message represents. All newly created data store
// contexts start as a local context on the client that created
// them, and for stashed ops, the client that applies it plays
// the role of creating client.
const dataStoreContext = new LocalFluidDataStoreContext({
id,
pkg: undefined,
parentContext: this.wrapContextForInnerChannel(id),
storage,
scope: this.parentContext.scope,
createSummarizerNodeFn: this.parentContext.getCreateChildSummarizerNodeFn(id, {
type: CreateSummarizerNodeSource.FromSummary,
}),
makeLocallyVisibleFn: () => this.makeDataStoreLocallyVisible(id),
snapshotTree,
});
// add to the list of bound or remoted, as this context must be bound
// to had an attach message sent, and is the non-detached case is remoted.
this.contexts.addBoundOrRemoted(dataStoreContext);
// realize the local context, as local contexts shouldn't be delay
// loaded, as this client is playing the role of creating client,
// and creating clients always create realized data store contexts.
const channel = await dataStoreContext.realize();
await channel.entryPoint.get();
if (this.parentContext.attachState !== AttachState.Detached) {
// if the client is not detached put in the pending attach list
// so that on ack of the stashed op, the context is found.
// detached client don't send ops, so should not expect and ack.
this.pendingAttach.set(id, message);
}
}
/**
* Process messages for this channel collection. The messages here are contiguous messages in a batch.
* @param messageCollection - The collection of messages to process.
*/
public processMessages(messageCollection: IRuntimeMessageCollection): void {
switch (messageCollection.envelope.type) {
case ContainerMessageType.FluidDataStoreOp: {
this.processChannelMessages(messageCollection);
break;
}
case ContainerMessageType.Attach: {
this.processAttachMessages(messageCollection);
break;
}
case ContainerMessageType.Alias: {
this.processAliasMessages(messageCollection);
break;
}
default: {
assert(false, 0x8e9 /* unreached */);
}
}
}
/**
* Process channel messages. The messages here are contiguous channel type messages in a batch. Bunch
* of contiguous messages for a data store should be sent to it together.
* @param messageCollection - The collection of messages to process.
*/
private processChannelMessages(messageCollection: IRuntimeMessageCollection): void {
const { envelope, messagesContent, local } = messageCollection;