Vihren Atlas

util.go ×30

Code frontier

unlabeled · c_a556dc62a970

44 tests · 4621 LOC · 185 files · introduces 0 tests · 266 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
40 ranges266 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
869 ranges4621 lines · 185 files · Browse complete extent
All tests (intent)
44 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

3 files ranked by introduced lines: 266 introduced LOC across 40 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/common/persistence/cassandra/util.go 218 introduced LOC · 30 ranges

Open complete file

266 shardID int32,
267 workflowSnapshot *p.InternalWorkflowSnapshot,
268 > ) error { util.go
269 > namespaceID := workflowSnapshot.NamespaceID
270 > workflowID := workflowSnapshot.WorkflowID
271 > runID := workflowSnapshot.RunID
272 >
273 > if err := createExecution(
274 > batch,
275 > shardID,
276 > workflowSnapshot,
277 > ); err != nil {
278 return err
279 }
280
281 > if err := updateActivityInfos( util.go
282 > batch,
283 > workflowSnapshot.ActivityInfos,
284 > nil,
285 > shardID,
286 > namespaceID,
287 > workflowID,
288 > runID,
289 > ); err != nil {
290 return err
291 }
292
293 > if err := updateTimerInfos( util.go
294 > batch,
295 > workflowSnapshot.TimerInfos,
296 > nil,
297 > shardID,
298 > namespaceID,
299 > workflowID,
300 > runID,
301 > ); err != nil {
302 return err
303 }
304
305 > if err := updateChildExecutionInfos( util.go
306 > batch,
307 > workflowSnapshot.ChildExecutionInfos,
308 > nil,
309 > shardID,
310 > namespaceID,
311 > workflowID,
312 > runID,
313 > ); err != nil {
314 return err
315 }
316
317 > if err := updateRequestCancelInfos( util.go
318 > batch,
319 > workflowSnapshot.RequestCancelInfos,
320 > nil,
321 > shardID,
322 > namespaceID,
323 > workflowID,
324 > runID,
325 > ); err != nil {
326 return err
327 }
328
329 > if err := updateSignalInfos( util.go
330 > batch,
331 > workflowSnapshot.SignalInfos,
332 > nil,
333 > shardID,
334 > namespaceID,
335 > workflowID,
336 > runID,
337 > ); err != nil {
338 return err
339 }
340
341 > if err := updateChasmNodes( util.go
342 > batch,
343 > workflowSnapshot.ChasmNodes,
344 > nil,
345 > shardID,
346 > namespaceID,
347 > workflowID,
348 > runID,
349 > ); err != nil {
350 return err
351 }
352
353 > updateSignalsRequested( util.go
354 > batch,
355 > workflowSnapshot.SignalRequestedIDs,
356 > nil,
357 > shardID,
358 > namespaceID,
359 > workflowID,
360 > runID,
361 > )
362 >
363 > // transfer / replication / timer tasks
364 > return applyTasks(
365 > batch,
366 > shardID,
367 > workflowSnapshot.Tasks,
368 > )
369 }
370
373 shardID int32,
374 snapshot *p.InternalWorkflowSnapshot,
375 > ) error { util.go
376 > // validate workflow state & close status
377 > if err := p.ValidateCreateWorkflowStateStatus(
378 > snapshot.ExecutionState.State,
379 > snapshot.ExecutionState.Status); err != nil {
380 return err
381 }
382
383 // TODO also need to set the start / current / last write version
384 > batch.Query(templateCreateWorkflowExecutionQuery, util.go
385 > shardID,
386 > snapshot.NamespaceID,
387 > snapshot.WorkflowID,
388 > snapshot.RunID,
389 > rowTypeExecution,
390 > snapshot.ExecutionInfoBlob.Data,
391 > snapshot.ExecutionInfoBlob.EncodingType.String(),
392 > snapshot.ExecutionStateBlob.Data,
393 > snapshot.ExecutionStateBlob.EncodingType.String(),
394 > snapshot.NextEventID,
395 > snapshot.DBRecordVersion,
396 > defaultVisibilityTimestamp,
397 > rowTypeExecutionTaskID,
398 > snapshot.Checksum.Data,
399 > snapshot.Checksum.EncodingType.String(),
400 > )
401 >
402 > return nil
403 }
404
616 workflowID string,
617 runID string,
618 > ) error { util.go
619 >
620 > for scheduledEventID, blob := range activityInfos {
621 > batch.Query(templateUpdateActivityInfoQuery,
622 > scheduledEventID,
623 > blob.Data,
624 > blob.EncodingType.String(),
625 > shardID,
626 > rowTypeExecution,
627 > namespaceID,
628 > workflowID,
629 > runID,
630 > defaultVisibilityTimestamp,
631 > rowTypeExecutionTaskID)
632 > }
633
634 > for deleteID := range deleteIDs { util.go
635 batch.Query(templateDeleteActivityInfoQuery,
636 deleteID,
643 rowTypeExecutionTaskID)
644 }
645 > return nil util.go
646 }
647
699 workflowID string,
700 runID string,
701 > ) error { util.go
702 > for timerID, blob := range timerInfos {
703 > batch.Query(templateUpdateTimerInfoQuery,
704 > timerID,
705 > blob.Data,
706 > blob.EncodingType.String(),
707 > shardID,
708 > rowTypeExecution,
709 > namespaceID,
710 > workflowID,
711 > runID,
712 > defaultVisibilityTimestamp,
713 > rowTypeExecutionTaskID)
714 > }
715
716 > for deleteInfoID := range deleteInfos { util.go
717 batch.Query(templateDeleteTimerInfoQuery,
718 deleteInfoID,
726 }
727
728 > return nil util.go
729 }
730
764 workflowID string,
765 runID string,
766 > ) error { util.go
767 >
768 > for initiatedId, blob := range childExecutionInfos {
769 > batch.Query(templateUpdateChildExecutionInfoQuery,
770 > initiatedId,
771 > blob.Data,
772 > blob.EncodingType.String(),
773 > shardID,
774 > rowTypeExecution,
775 > namespaceID,
776 > workflowID,
777 > runID,
778 > defaultVisibilityTimestamp,
779 > rowTypeExecutionTaskID)
780 > }
781
782 > for deleteID := range deleteIDs { util.go
783 batch.Query(templateDeleteChildExecutionInfoQuery,
784 deleteID,
791 rowTypeExecutionTaskID)
792 }
793 > return nil util.go
794 }
795
829 workflowID string,
830 runID string,
831 > ) error { util.go
832 >
833 > for initiatedId, blob := range requestCancelInfos {
834 > batch.Query(templateUpdateRequestCancelInfoQuery,
835 > initiatedId,
836 > blob.Data,
837 > blob.EncodingType.String(),
838 > shardID,
839 > rowTypeExecution,
840 > namespaceID,
841 > workflowID,
842 > runID,
843 > defaultVisibilityTimestamp,
844 > rowTypeExecutionTaskID)
845 > }
846
847 > for deleteID := range deleteIDs { util.go
848 batch.Query(templateDeleteRequestCancelInfoQuery,
849 deleteID,
856 rowTypeExecutionTaskID)
857 }
858 > return nil util.go
859 }
860
894 workflowID string,
895 runID string,
896 > ) error { util.go
897 >
898 > for initiatedId, blob := range signalInfos {
899 > batch.Query(templateUpdateSignalInfoQuery,
900 > initiatedId,
901 > blob.Data,
902 > blob.EncodingType.String(),
903 > shardID,
904 > rowTypeExecution,
905 > namespaceID,
906 > workflowID,
907 > runID,
908 > defaultVisibilityTimestamp,
909 > rowTypeExecutionTaskID)
910 > }
911
912 > for deleteID := range deleteIDs { util.go
913 batch.Query(templateDeleteSignalInfoQuery,
914 deleteID,
921 rowTypeExecutionTaskID)
922 }
923 > return nil util.go
924 }
925
988 workflowID string,
989 runID string,
990 > ) error { util.go
991 > for deletePath := range deleteNodes {
992 batch.Query(templateDeleteChasmNodeQuery,
993 deletePath,
1001 }
1002
1003 > for upsertPath, node := range upsertNodes { util.go
1004 > batch.Query(templateUpdateChasmNodeQuery,
1005 > upsertPath,
1006 > node.CassandraBlob.Data,
1007 > node.CassandraBlob.EncodingType.String(),
1008 > shardID,
1009 > rowTypeExecution,
1010 > namespaceID,
1011 > workflowID,
1012 > runID,
1013 > defaultVisibilityTimestamp,
1014 > rowTypeExecutionTaskID)
1015 > }
1016
1017 > return nil util.go
1018 }
1019
1026 workflowID string,
1027 runID string,
1028 > ) { util.go
1029 >
1030 > if len(signalReqIDs) > 0 {
1031 > batch.Query(templateUpdateSignalRequestedQuery,
1032 > convert.StringSetToSlice(signalReqIDs),
1033 > shardID,
1034 > rowTypeExecution,
1035 > namespaceID,
1036 > workflowID,
1037 > runID,
1038 > defaultVisibilityTimestamp,
1039 > rowTypeExecutionTaskID)
1040 > }
1041
1042 > if len(deleteSignalReqIDs) > 0 { util.go
1043 batch.Query(templateDeleteWorkflowExecutionSignalRequestedQuery,
1044 convert.StringSetToSlice(deleteSignalReqIDs),
go.temporal.io/server/common/persistence/cassandra/mutable_state_store.go 45 introduced LOC · 8 ranges

Open complete file

384 ctx context.Context,
385 request *p.InternalCreateWorkflowExecutionRequest,
386 > ) (*p.InternalCreateWorkflowExecutionResponse, error) { mutable_state_store.go
387 > batch := d.Session.NewBatch(gocql.LoggedBatch).WithContext(ctx)
388 >
389 > shardID := request.ShardID
390 > newWorkflow := request.NewWorkflowSnapshot
391 > lastWriteVersion := newWorkflow.LastWriteVersion
392 > namespaceID := newWorkflow.NamespaceID
393 > workflowID := newWorkflow.WorkflowID
394 > runID := newWorkflow.RunID
395 >
396 > var requestCurrentRunID string
397 > currentRecordRunID := d.getCurrentRecordRunID(request.ArchetypeID)
398 >
399 > switch request.Mode {
400 case p.CreateWorkflowModeBypassCurrent:
401 // noop
444 }
445
446 > if err := applyWorkflowSnapshotBatchAsNew(batch, mutable_state_store.go
447 > request.ShardID,
448 > &newWorkflow,
449 > ); err != nil {
450 return nil, err
451 }
452
453 > batch.Query(templateUpdateLeaseQuery, mutable_state_store.go
454 > request.RangeID,
455 > request.ShardID,
456 > rowTypeShard,
457 > rowTypeShardNamespaceID,
458 > rowTypeShardWorkflowID,
459 > rowTypeShardRunID,
460 > defaultVisibilityTimestamp,
461 > rowTypeShardTaskID,
462 > request.RangeID,
463 > )
464 >
465 > conflictRecord := newConflictRecord()
466 > applied, conflictIter, err := d.Session.MapExecuteBatchCAS(batch, conflictRecord)
467 > if err != nil {
468 return nil, gocql.ConvertError("CreateWorkflowExecution", err)
469 }
470 > defer func() { mutable_state_store.go
471 > _ = conflictIter.Close()
472 > }()
473
474 > if !applied { mutable_state_store.go
475 return nil, convertErrors(
476 conflictRecord,
1092 func (d *MutableStateStore) getCurrentRecordRunID(
1093 archetypeID chasm.ArchetypeID,
1094 > ) string { mutable_state_store.go
1095 > if !softassert.That(
1096 > d.logger,
1097 > archetypeID != chasm.UnspecifiedArchetypeID,
1098 > "ArchetypeID not specified, defaulting to Workflow.",
1099 > ) {
1100 return permanentRunID
1101 }
1102
1103 > if archetypeID == chasm.WorkflowArchetypeID { mutable_state_store.go
1104 return permanentRunID
1105 }
go.temporal.io/server/common/persistence/cassandra/execution_store.go 3 introduced LOC · 2 ranges

Open complete file

98 ctx context.Context,
99 request *p.InternalCreateWorkflowExecutionRequest,
100 > ) (*p.InternalCreateWorkflowExecutionResponse, error) { execution_store.go
101 > for _, req := range request.NewWorkflowNewEvents {
102 if err := d.AppendHistoryNodes(ctx, req); err != nil {
103 return nil, err
105 }
106
107 > return d.MutableStateStore.CreateWorkflowExecution(ctx, request) execution_store.go
108 }
109

Direct hierarchy

Parents

Children