TestReplaysWithDynamicConfigChange

go.temporal.io/server/service/worker/scheduler/workflow.go 833 covered LOC · 220 ranges

Open complete file

180 )

181

182 > func defaultLocalActivityOptions() workflow.LocalActivityOptions { workflow.go

183 > return workflow.LocalActivityOptions{

184 > // This applies to watch, cancel, and terminate. Start workflow overrides this.

185 > ScheduleToCloseTimeout: 1 * time.Hour,

186 > // Each local activity is one or a few local RPCs.

187 > // Currently this is applied manually, see https://github.com/temporalio/sdk-go/issues/1066

188 > StartToCloseTimeout: 5 * time.Second,

189 > RetryPolicy: &temporal.RetryPolicy{

190 > InitialInterval: 1 * time.Second,

191 > MaximumInterval: 60 * time.Second,

192 > },

193 > }

194 > }

195

196 var (

239 }

240

241 > func schedulerWorkflowWithSpecBuilder(ctx workflow.Context, args *schedulespb.StartScheduleArgs, specBuilder *SpecBuilder, enableCHASMMigration func() bool, migrateWithRunningWorkflows func() bool) error { workflow.go

242 > scheduler := &scheduler{

243 > StartScheduleArgs: args,

244 > ctx: ctx,

245 > a: nil,

246 > logger: sdklog.With(workflow.GetLogger(ctx), "wf-namespace", args.State.Namespace, "schedule-id", args.State.ScheduleId),

247 > metrics: workflow.GetMetricsHandler(ctx).WithTags(map[string]string{

248 > "namespace": args.State.Namespace,

249 > metrics.ScheduleBackendTag: metrics.ScheduleBackendLegacy,

250 > }),

251 > specBuilder: specBuilder,

252 > enableCHASMMigration: enableCHASMMigration,

253 > migrateWithRunningWorkflows: migrateWithRunningWorkflows,

254 > }

255 > return scheduler.run()

256 > }

257

258 > func (s *scheduler) run() error { workflow.go

259 > s.updateTweakables()

260 > s.ensureFields()

261 > s.compileSpec()

262 >

263 > if err := workflow.SetQueryHandler(s.ctx, QueryNameDescribe, s.handleDescribeQuery); err != nil {

264 return err

265 }

266 > if err := workflow.SetQueryHandler(s.ctx, QueryNameListMatchingTimes, s.handleListMatchingTimesQuery); err != nil { workflow.go

267 return err

268 }

269

270 > if s.State.LastProcessedTime == nil { workflow.go

271 > // log these as json since it's more readable than the Go representation workflow.go

272 > specJson, _ := protojson.Marshal(s.Schedule.Spec)

273 > policiesJson, _ := protojson.Marshal(s.Schedule.Policies)

274 > s.logger.Info("Starting schedule", "spec", string(specJson), "policies", string(policiesJson))

275 >

276 > s.State.LastProcessedTime = timestamppb.New(s.now())

277 > s.State.ConflictToken = InitialConflictToken

278 > s.Info.CreateTime = s.State.LastProcessedTime

279 >

280 > s.recordActionPayloadMetrics()

281 > }

282

283 // A schedule may be created with an initial Patch, e.g. start one immediately. Put that in

284 // the state so it takes effect below.

285 > s.pendingPatch = s.InitialPatch workflow.go

286 > s.InitialPatch = nil

287 >

288 > iters := s.tweakables.IterationsBeforeContinueAsNew

289 > for {

290 > info := workflow.GetInfo(s.ctx)

291 > suggestContinueAsNew := info.GetCurrentHistoryLength() >= impossibleHistorySize

292 > if s.tweakables.IterationsBeforeContinueAsNew > 0 {

293 > // forceCAN must be checked here too, not just in the else branch, workflow.go

294 > // so that the force-continue-as-new signal is honored regardless

295 > // of whether IterationsBeforeContinueAsNew is set.

296 > suggestContinueAsNew = suggestContinueAsNew || iters <= 0 || s.forceCAN

297 > iters--

298 > } else { workflow.go

299 > suggestContinueAsNew = suggestContinueAsNew || info.GetContinueAsNewSuggested() || s.forceCAN workflow.go

300 > }

301 > if suggestContinueAsNew && s.pendingUpdate == nil && s.pendingPatch == nil { workflow.go

302 break

303 }

304

305 > t1 := timestamp.TimeValue(s.State.LastProcessedTime) workflow.go

306 > t2 := s.now()

307 > if t2.Before(t1) {

308 // Time went backwards. Currently this can only happen across a continue-as-new boundary.

309 s.logger.Warn("Time went backwards", "from", t1, "to", t2)

310 t2 = t1

311 }

312 > nextWakeup, lastAction := s.processTimeRange( workflow.go

313 > t1, t2,

314 > // resolve this to the schedule's policy as late as possible

315 > enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,

316 > false,

317 > nil,

318 > )

319 > if s.hasMinVersion(UseLastAction) {

320 > s.State.LastProcessedTime = timestamppb.New(lastAction)

321 > } else {

322 > s.State.LastProcessedTime = timestamppb.New(t2) workflow.go

323 > }

324 // handle signals after processing time range that just elapsed

325 > scheduleChanged := s.processSignals() workflow.go

326 > if scheduleChanged {

327 > // need to calculate sleep again workflow.go

328 > nextWakeup, _ = s.processTimeRange(

329 > s.State.LastProcessedTime.AsTime(),

330 > t2,

331 > enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED,

332 > false,

333 > nil,

334 > )

335 > }

336

337 > if !s.State.PendingMigration && s.tweakables.EnableCHASMMigration && workflow.go

338 > (s.tweakables.MigrateWithRunningWorkflows || len(s.Info.RunningWorkflows) == 0) {

339 s.State.PendingMigration = true

340 }

341 > if s.State.PendingMigration { workflow.go

342 err := s.executeMigration()

343 if err == nil {

362

363 // process backfills if we have any too

364 > s.processBackfills() workflow.go

365 > // try starting workflows in the buffer

366 > //nolint:revive

367 > for s.processBuffer() {

368 > } workflow.go

369 > s.updateMemoAndSearchAttributes() workflow.go

370 >

371 > // if schedule is not paused and out of actions or do not have anything scheduled, exit the schedule workflow after retention period has passed

372 > if exp := s.getRetentionExpiration(nextWakeup); !exp.IsZero() && !exp.After(s.now()) {

373 return nil

374 }

375 > if suggestContinueAsNew && s.pendingUpdate == nil && s.pendingPatch == nil && s.hasMinVersion(CANAfterSignals) { workflow.go

376 // If suggestContinueAsNew was true but we had a pending update or patch, we would

377 // not break above, but process the update/patch. Now that we're done, we should

385 // 2. we got a signal (update, request, refresh)

386 // 3. a workflow that we were watching finished

387 > s.sleep(nextWakeup) workflow.go

388 > s.updateTweakables()

389 }

390

395 }

396

397 > func (s *scheduler) ensureFields() { workflow.go

398 > if s.Schedule == nil {

399 s.Schedule = &schedulepb.Schedule{}

400 }

401 > if s.Schedule.Spec == nil { workflow.go

402 s.Schedule.Spec = &schedulepb.ScheduleSpec{}

403 }

404 > if s.Schedule.Action == nil { workflow.go

405 s.Schedule.Action = &schedulepb.ScheduleAction{}

406 }

407 > if s.Schedule.Policies == nil { workflow.go

408 > s.Schedule.Policies = &schedulepb.SchedulePolicies{} workflow.go

409 > }

410

411 // set defaults eagerly so they show up in describe output

412 > s.Schedule.Policies.OverlapPolicy = s.resolveOverlapPolicy(s.Schedule.Policies.OverlapPolicy) workflow.go

413 > s.Schedule.Policies.CatchupWindow = durationpb.New(s.getCatchupWindow())

414 >

415 > if s.Schedule.State == nil {

416 > s.Schedule.State = &schedulepb.ScheduleState{} workflow.go

417 > }

418 > if s.Info == nil { workflow.go

419 > s.Info = &schedulepb.ScheduleInfo{}

420 > }

421 > if s.State == nil {

422 s.State = &schedulespb.InternalState{}

423 }

424 }

425

426 > func (s *scheduler) compileSpec() { workflow.go

427 > // if spec changes invalidate current cache

428 > s.nextTimeCacheV1 = nil

429 > s.nextTimeCacheV2 = nil

430 >

431 > cspec, err := s.specBuilder.NewCompiledSpec(s.Schedule.Spec)

432 > if err != nil {

433 if s.logger != nil {

434 s.logger.Error("Invalid schedule", "error", err)

436 s.Info.InvalidScheduleError = err.Error()

437 s.cspec = nil

438 > } else { workflow.go

439 > s.Info.InvalidScheduleError = ""

440 > s.cspec = cspec

441 > }

442 }

443

444 > func (s *scheduler) now() time.Time { workflow.go

445 > // Notes:

446 > // 1. The time returned here is actually the timestamp of the WorkflowTaskStarted

447 > // event, which is generated in history, not any time on the worker itself.

448 > // 2. There will be some delay between when history stamps the time on the

449 > // WorkflowTaskStarted event and when this code runs, as the event+task goes through

450 > // matching and frontend. But it should be well under a second, which is our minimum

451 > // granularity anyway.

452 > // 3. It's actually the maximum of all of those events: the go sdk enforces that

453 > // workflow time is monotonic. So if the clock on a history node is wrong and workflow

454 > // time is ahead of real time for a while, and then goes back, scheduled jobs will run

455 > // ahead of time, and then nothing will happen until real time catches up to where it

456 > // was temporarily ahead. Currently the only way to "recover" from this situation is

457 > // to recreate the schedule/scheduler workflow.

458 > // 4. Actually there is one way time can appear to go backwards from the point of view

459 > // of this workflow: across a continue-as-new, since monotonicity isn't preserved

460 > // there (as far as I know). We'll treat that the same since we keep track of the last

461 > // processed schedule time.

462 > return workflow.Now(s.ctx)

463 > }

464

465 > func (s *scheduler) processPatch(patch *schedulepb.SchedulePatch) { workflow.go

466 > s.logger.Debug("Schedule patch")

467 >

468 > if trigger := patch.TriggerImmediately; trigger != nil {

469 > now := s.now() workflow.go

470 > if s.hasMinVersion(TriggerImmediatelyTimestamp) {

471 now = timestamp.TimeValue(trigger.ScheduledTime)

472 if now.IsZero() {

474 }

475 }

476 > s.addStart(now, now, trigger.OverlapPolicy, true) workflow.go

477 }

478

479 > for _, bfr := range patch.BackfillRequest { workflow.go

480 > // In previous versions the backfill start time was exclusive, ie when workflow.go

481 > // the start time of the backfill matched the schedule's spec, it would

482 > // not be executed. This new version makes it inclusive instead.

483 > if s.hasMinVersion(InclusiveBackfillStartTime) {

484 > startTime := timestamp.TimeValue(bfr.GetStartTime()).Add(-1 * time.Millisecond)

485 > bfr.StartTime = timestamppb.New(startTime)

486 > }

487 > if s.hasMinVersion(IncrementalBackfill) {

488 > // Add to ongoing backfills to process incrementally

489 > if len(s.State.OngoingBackfills) >= s.tweakables.MaxBufferSize {

490 s.logger.Warn("Buffer overrun for backfill requests")

491 s.metrics.Counter(metrics.ScheduleBufferOverruns.Name()).Inc(1)

493 continue

494 }

495 > s.State.OngoingBackfills = append(s.State.OngoingBackfills, common.CloneProto(bfr)) workflow.go

496 > } else { workflow.go

497 > // Old version: process whole backfill synchronously

498 > s.processTimeRange(

499 > timestamp.TimeValue(bfr.GetStartTime()),

500 > timestamp.TimeValue(bfr.GetEndTime()),

501 > bfr.GetOverlapPolicy(),

502 > true,

503 > nil,

504 > )

505 > }

506 }

507

508 > if patch.Pause != "" { workflow.go

509 > s.Schedule.State.Paused = true workflow.go

510 > s.Schedule.State.Notes = patch.Pause

511 > s.incSeqNo()

512 > }

513 > if patch.Unpause != "" { workflow.go

514 > s.Schedule.State.Paused = false workflow.go

515 > s.Schedule.State.Notes = patch.Unpause

516 > s.incSeqNo()

517 > }

518 }

519

520 > func (s *scheduler) getNextTimeV1(after time.Time) GetNextTimeResult { workflow.go

521 > // we populate the map sequentially, if after is not in the map, it means we either exhausted

522 > // all items, or we jumped through time (forward or backward), in either case, refresh the cache

523 > next, ok := s.nextTimeCacheV1[after]

524 > if ok {

525 > return next

526 > }

527 > s.nextTimeCacheV1 = nil

528 > // Run this logic in a SideEffect so that we can fix bugs there without breaking

529 > // existing schedule workflows.

530 > panicIfErr(workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {

531 results := make(map[time.Time]GetNextTimeResult)

532 for t := after; !t.IsZero() && len(results) < nextTimeCacheV1Size; {

539 return results

540 }).Get(&s.nextTimeCacheV1))

541 > return s.nextTimeCacheV1[after] workflow.go

542 }

543

545 // refilled, try to refill it starting from `cacheBase` instead of `after`. This avoids having

546 // the cache range jump back and forth when generating a sequence of times.

547 > func (s *scheduler) getNextTimeV2(cacheBase, after time.Time) GetNextTimeResult { workflow.go

548 > // cacheBase must be before after

549 > cacheBase = util.MinTime(cacheBase, after)

550 >

551 > // Asking for a time before the cache, need to refill.

552 > // Also if version changed (so we can fix a bug immediately).

553 > if s.nextTimeCacheV2 == nil ||

554 > after.Before(s.nextTimeCacheV2.StartTime.AsTime()) ||

555 > SchedulerWorkflowVersion(s.nextTimeCacheV2.Version) != s.tweakables.Version {

556 > s.fillNextTimeCacheV2(cacheBase)

557 > }

558

559 // We may need up to three tries: the first is in the cache as it exists now,

560 // the second is refilled from cacheBase, and the third is if cacheBase was set

561 // too far in the past, we ignore it and fill the cache from after.

562 > for try := 1; try <= 3; try++ { workflow.go

563 > if res, ok := searchCache(s.nextTimeCacheV2, after); ok {

564 > return res

565 > }

566 // Otherwise refill from base

567 > s.fillNextTimeCacheV2(cacheBase) workflow.go

568 > // Reset cacheBase so that if it was set too early and we run off the end again, the

569 > // third try will work.

570 > cacheBase = after

571 }

572

576 }

577

578 > func searchCache(cache *schedulespb.NextTimeCache, after time.Time) (GetNextTimeResult, bool) { workflow.go

579 > // The cache covers a contiguous time range so we can do a linear search in it.

580 > start := cache.StartTime.AsTime()

581 > afterOffset := int64(after.Sub(start))

582 > for i, nextOffset := range cache.NextTimes {

583 > if nextOffset > afterOffset { workflow.go

584 > next := start.Add(time.Duration(nextOffset))

585 > nominal := next

586 > if i < len(cache.NominalTimes) && cache.NominalTimes[i] != 0 {

587 > nominal = start.Add(time.Duration(cache.NominalTimes[i])) workflow.go

588 > }

589 > return GetNextTimeResult{Nominal: nominal, Next: next}, true workflow.go

590 }

591 }

592 // Ran off end: if completed, then we're done

593 > if cache.Completed { workflow.go

594 > return GetNextTimeResult{}, true workflow.go

595 > }

596 > return GetNextTimeResult{}, false workflow.go

597 }

598

599 > func (s *scheduler) fillNextTimeCacheV2(start time.Time) { workflow.go

600 > // Clear value so we can Get into it

601 > s.nextTimeCacheV2 = nil

602 > // Run this logic in a SideEffect so that we can fix bugs there without breaking

603 > // existing schedule workflows.

604 > val := workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {

605 cache := &schedulespb.NextTimeCache{

606 Version: int64(s.tweakables.Version),

641 // but it might not. To be sure, check StartTime also (the field names differ between json

642 // and proto so json.Unmarshal will never fill in StartTime).

643 > if val.Get(&s.nextTimeCacheV2) == nil && s.nextTimeCacheV2.GetStartTime() != nil { workflow.go

644 > return

645 > }

646 // Try as json value

647 > var jsonVal jsonNextTimeCacheV2 workflow.go

648 > if val.Get(&jsonVal) != nil || jsonVal.Start.IsZero() {

649 panic("could not decode next time cache as proto or json")

650 }

651 > s.nextTimeCacheV2 = &schedulespb.NextTimeCache{ workflow.go

652 > Version: int64(jsonVal.Version),

653 > StartTime: timestamppb.New(jsonVal.Start),

654 > NextTimes: make([]int64, len(jsonVal.Results)),

655 > NominalTimes: make([]int64, len(jsonVal.Results)),

656 > Completed: jsonVal.Completed,

657 > }

658 > for i, res := range jsonVal.Results {

659 > s.nextTimeCacheV2.NextTimes[i] = int64(res.Next.Sub(jsonVal.Start))

660 > s.nextTimeCacheV2.NominalTimes[i] = int64(res.Nominal.Sub(jsonVal.Start))

661 > }

662 }

663

664 > func (s *scheduler) getNextTime(after time.Time) GetNextTimeResult { workflow.go

665 > // Implementation using a cache to save markers + computation.

666 > if s.hasMinVersion(NewCacheAndJitter) {

667 > return s.getNextTimeV2(after, after)

668 > } else if s.hasMinVersion(BatchAndCacheTimeQueries) {

669 > return s.getNextTimeV1(after) workflow.go

670 > }

671 // Run this logic in a SideEffect so that we can fix bugs there without breaking

672 // existing schedule workflows.

673 > var next GetNextTimeResult workflow.go

674 > panicIfErr(workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {

675 res, _ := s.cspec.GetNextTime(s.jitterSeed(), after)

676 return res

677 }).Get(&next))

678 > return next workflow.go

679 }

680

684 manual bool,

685 limit *int,

686 > ) (time.Time, time.Time) { workflow.go

687 > s.logger.Debug("processTimeRange", "start", start, "end", end, "overlap-policy", overlapPolicy, "manual", manual)

688 >

689 > if s.cspec == nil {

690 return time.Time{}, end

691 }

692

693 > catchupWindow := s.getCatchupWindow() workflow.go

694 >

695 > // A previous version would record a marker for each time which could make a workflow

696 > // fail. With the new version, the entire time range is skipped if the workflow is paused

697 > // or we are not going to take an action now

698 > if s.hasMinVersion(BatchAndCacheTimeQueries) {

699 > // Peek at paused/remaining actions state and don't bother if we're not going to

700 > // take an action now. (Don't count as missed catchup window either.)

701 > // Skip over entire time range if paused or no actions can be taken

702 > if !s.canTakeScheduledAction(manual, false) {

703 > // use end as last action time so that we don't reprocess time spent paused workflow.go

704 > return s.getNextTime(end).Next, end

705 > }

706 }

707

708 > lastAction := start workflow.go

709 > recordedGenerateLatency := false

710 > var next GetNextTimeResult

711 > for next = s.getNextTime(start); !(next.Next.IsZero() || next.Next.After(end)); next = s.getNextTime(next.Next) {

712 > if !s.hasMinVersion(BatchAndCacheTimeQueries) && !s.canTakeScheduledAction(manual, false) { workflow.go

713 > continue workflow.go

714 }

715 > if !manual && s.Info.UpdateTime.AsTime().After(next.Next) { workflow.go

716 // We're reprocessing since the most recent event after an update. Discard actions before

717 // the update time (which was just set to "now"). This doesn't have to be guarded with

721 // Record generate latency only for the first action in the batch to

722 // avoid inflating the metric when catching up over a large time range.

723 > if !manual && !recordedGenerateLatency { workflow.go

724 > s.metrics.Timer(metrics.ScheduleGenerateLatency.Name()).Record(end.Sub(next.Next)) workflow.go

725 > recordedGenerateLatency = true

726 > }

727 > if !manual && end.Sub(next.Next) > catchupWindow { workflow.go

728 s.logger.Warn("Schedule missed catchup window", "now", end, "time", next.Next)

729 // Action's nominal time was already past the catchup window when

737 continue

738 }

739 > s.addStart(next.Nominal, next.Next, overlapPolicy, manual) workflow.go

740 > lastAction = next.Next

741 >

742 > if limit != nil {

743 > if (*limit)--; *limit <= 0 { workflow.go

744 break

745 }

746 }

747 }

748 > return next.Next, lastAction workflow.go

749 }

750

751 > func (s *scheduler) canTakeScheduledAction(manual, decrement bool) bool { workflow.go

752 > // If manual (trigger immediately or backfill), always allow

753 > if manual {

754 > return true workflow.go

755 > }

756 // If paused, don't do anything

757 > if s.Schedule.State.Paused { workflow.go

758 > return false workflow.go

759 > }

760 // If unlimited actions, allow

761 > if !s.Schedule.State.LimitedActions { workflow.go

762 > return true workflow.go

763 > }

764 // Otherwise check and decrement limit

765 > if s.Schedule.State.RemainingActions > 0 { workflow.go

766 > if decrement {

767 > s.Schedule.State.RemainingActions--

768 > s.incSeqNo()

769 > }

770 > return true

771 }

772 // No actions left

773 > return false workflow.go

774 }

775

776 > func (s *scheduler) sleep(nextWakeup time.Time) { workflow.go

777 > sel := workflow.NewSelector(s.ctx)

778 >

779 > upCh := workflow.GetSignalChannel(s.ctx, SignalNameUpdate)

780 > sel.AddReceive(upCh, func(ch workflow.ReceiveChannel, _ bool) {

781 > ch.Receive(s.ctx, &s.pendingUpdate) workflow.go

782 > })

783

784 > reqCh := workflow.GetSignalChannel(s.ctx, SignalNamePatch) workflow.go

785 > sel.AddReceive(reqCh, func(ch workflow.ReceiveChannel, _ bool) {

786 > ch.Receive(s.ctx, &s.pendingPatch) workflow.go

787 > })

788

789 > refreshCh := workflow.GetSignalChannel(s.ctx, SignalNameRefresh) workflow.go

790 > sel.AddReceive(refreshCh, s.handleRefreshSignal)

791 >

792 > forceCAN := workflow.GetSignalChannel(s.ctx, SignalNameForceCAN)

793 > sel.AddReceive(forceCAN, s.handleForceCANSignal)

794 >

795 > migrateCh := workflow.GetSignalChannel(s.ctx, SignalNameMigrateToChasm)

796 > sel.AddReceive(migrateCh, s.handleMigrateSignal)

797 >

798 > if s.hasMoreAllowAllBackfills() {

799 // if we have more allow-all backfills to do, do a short sleep and continue

800 nextWakeup = s.now().Add(1 * time.Second)

801 > } else if s.tweakables.SleepWhilePaused && !s.canTakeScheduledAction(false, false) { workflow.go

802 > // if we're paused or out of actions, we don't need to wake up until we get an update workflow.go

803 > nextWakeup = time.Time{}

804 > }

805

806 // if retention is not zero, it means there is no more job to schedule, so sleep for retention time and then exit if no signal is received

807 > if exp := s.getRetentionExpiration(nextWakeup); !exp.IsZero() { workflow.go

808 > nextWakeup = exp workflow.go

809 > }

810

811 > if !nextWakeup.IsZero() { workflow.go

812 > sleepTime := nextWakeup.Sub(s.now()) workflow.go

813 > // A previous version of this workflow passed around sleep duration instead of wakeup time,

814 > // which means it always set a timer even in cases where sleepTime comes out negative. For

815 > // compatibility, we have to continue setting a positive timer in those cases. The value

816 > // doesn't have to match, though.

817 > if !s.tweakables.AllowZeroSleep && sleepTime <= 0 {

818 sleepTime = time.Second

819 }

820 // A previous version of this workflow always created a new timer here, which is wasteful.

821 // We can reuse a previous timer if we have the same deadline and it didn't fire yet.

822 > if s.tweakables.ReuseTimer { workflow.go

823 > if s.currentTimer == nil || !s.currentTimerDeadline.Equal(nextWakeup) {

824 > s.currentTimer = workflow.NewTimer(s.ctx, sleepTime)

825 > s.currentTimerDeadline = nextWakeup

826 > }

827 > sel.AddFuture(s.currentTimer, func(_ workflow.Future) {

828 > s.currentTimer = nil workflow.go

829 > })

830 > } else { workflow.go

831 > tmr := workflow.NewTimer(s.ctx, sleepTime)

832 > sel.AddFuture(tmr, func(_ workflow.Future) {})

833 }

834 }

835

836 > if s.watchingFuture != nil { workflow.go

837 > sel.AddFuture(s.watchingFuture, s.wfWatcherReturned) workflow.go

838 > }

839

840 > s.logger.Debug("sleeping", "next-wakeup", nextWakeup, "watching", s.watchingFuture != nil) workflow.go

841 > sel.Select(s.ctx)

842 > for sel.HasPending() {

843 > sel.Select(s.ctx) workflow.go

844 > }

845 }

846

847 > func (s *scheduler) processBackfills() { workflow.go

848 > limit := s.tweakables.BackfillsPerIteration

849 >

850 > for len(s.State.OngoingBackfills) > 0 &&

851 > limit > 0 &&

852 > // use only half the buffer for backfills

853 > len(s.State.BufferedStarts) < s.tweakables.MaxBufferSize/2 {

854 > bfr := s.State.OngoingBackfills[0] workflow.go

855 > startTime := timestamp.TimeValue(bfr.GetStartTime())

856 > endTime := timestamp.TimeValue(bfr.GetEndTime())

857 > next, _ := s.processTimeRange(

858 > startTime,

859 > endTime,

860 > bfr.GetOverlapPolicy(),

861 > true,

862 > &limit,

863 > )

864 > if next.IsZero() || next.After(endTime) {

865 > // done with this one

866 > s.State.OngoingBackfills = s.State.OngoingBackfills[1:]

867 > } else {

868 // adjust start time for next iteration

869 bfr.StartTime = timestamppb.New(next)

872 }

873

874 > func (s *scheduler) hasMoreAllowAllBackfills() bool { workflow.go

875 > return len(s.State.OngoingBackfills) > 0 &&

876 > s.resolveOverlapPolicy(s.State.OngoingBackfills[0].OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL

877 > }

878

879 > func (s *scheduler) wfWatcherReturned(f workflow.Future) { workflow.go

880 > id := s.watchingWorkflowId

881 > s.watchingWorkflowId = ""

882 > s.watchingFuture = nil

883 > s.processWatcherResult(id, f, true)

884 > }

885

886 > func (s *scheduler) processWatcherResult(id string, f workflow.Future, long bool) { workflow.go

887 > var res schedulespb.WatchWorkflowResponse

888 > err := f.Get(s.ctx, &res)

889 > if err != nil {

890 s.logger.Error("error from workflow watcher future", "workflow", id, "error", err, "long", long)

891 return

892 }

893

894 > if res.Status == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING { workflow.go

895 > // this should only happen for a refresh, not a long-poll workflow.go

896 > if long {

897 s.logger.Warn("watcher returned for running workflow", "workflow", id)

898 }

899 > return workflow.go

900 }

901

902 // now we know it's not running, remove from running workflow list

903 > matchRunning := func(ex *commonpb.WorkflowExecution) bool { return ex.GetWorkflowId() == id } workflow.go

904 > if idx := slices.IndexFunc(s.Info.RunningWorkflows, matchRunning); idx >= 0 {

905 > // We could also immediately update visibility, with updateMemoAndSearchAttributes,

906 > // but the wakeup here will trigger a write after anyways (after processing the buffer).

907 > s.Info.RunningWorkflows = slices.Delete(s.Info.RunningWorkflows, idx, idx+1)

908 > } else {

909 // This could happen if the watcher activity gets interrupted and is retried after the

910 // heartbeat timeout, but in the meantime we have done a refresh through a local activity.

914

915 // update workflow execution status in RecentActions

916 > if s.hasMinVersion(ActionResultIncludesStatus) { workflow.go

917 matchRecent := func(a *schedulepb.ScheduleActionResult) bool { return a.GetStartWorkflowResult().GetWorkflowId() == id }

918 if idx := slices.IndexFunc(s.Info.RecentActions, matchRecent); idx >= 0 {

922

923 // handle pause-on-failure

924 > failedStatus := res.Status == enumspb.WORKFLOW_EXECUTION_STATUS_FAILED || workflow.go

925 > res.Status == enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT ||

926 > (s.tweakables.CanceledTerminatedCountAsFailures &&

927 > (res.Status == enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED || res.Status == enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED))

928 > pauseOnFailure := s.Schedule.Policies.PauseOnFailure && failedStatus && !s.Schedule.State.Paused

929 > if pauseOnFailure {

930 s.Schedule.State.Paused = true

931 if res.Status == enumspb.WORKFLOW_EXECUTION_STATUS_FAILED {

940

941 // handle last completion/failure and record resulting payload sizes

942 > if res.GetResult() != nil { workflow.go

943 s.State.LastCompletionResult = res.GetResult()

944 s.State.ContinuedFailure = nil

947 s.metrics.Counter(metrics.SchedulePayloadSize.Name()).Inc(int64(len(resultPayload)))

948 }

949 > } else if res.GetFailure() != nil { workflow.go

950 // leave LastCompletionResult from previous run

951 s.State.ContinuedFailure = res.GetFailure()

961 // (Attempt == -1) like CHASM, so BufferedStarts[0] is always the next

962 // pending start.

963 > if long && len(s.State.BufferedStarts) > 0 { workflow.go

964 > s.State.BufferedStarts[0].DesiredTime = res.CloseTime workflow.go

965 > }

966

967 > s.logger.Debug("started workflow finished", "workflow", id, "status", res.Status, "pause-after-failure", pauseOnFailure, "long", long) workflow.go

968 }

969

973 // Payloads for workflow completions and failures are recorded as part of the

974 // same metric.

975 > func (s *scheduler) recordActionPayloadMetrics() { workflow.go

976 > startWf := s.Schedule.Action.GetStartWorkflow()

977 > if startWf == nil {

978 return

979 }

980

981 > inputPayload, err := startWf.Input.Marshal() workflow.go

982 > if err != nil {

983 return

984 }

985

986 > memoPayload, err := startWf.Memo.Marshal() workflow.go

987 > if err != nil {

988 return

989 }

990

991 > payloadSize := int64(len(memoPayload) + len(inputPayload)) workflow.go

992 > s.metrics.Counter(metrics.SchedulePayloadSize.Name()).Inc(payloadSize)

993 }

994

995 > func (s *scheduler) processUpdate(req *schedulespb.FullUpdateRequest) { workflow.go

996 > if err := s.checkConflict(req.ConflictToken); err != nil {

997 s.logger.Warn("Update conflicted with concurrent change")

998 return

999 }

1000

1001 > s.logger.Debug("Schedule update") workflow.go

1002 >

1003 > s.Schedule.Spec = req.Schedule.GetSpec()

1004 > s.Schedule.Action = req.Schedule.GetAction()

1005 > s.Schedule.Policies = req.Schedule.GetPolicies()

1006 > s.Schedule.State = req.Schedule.GetState()

1007 > // don't touch Info

1008 >

1009 > s.ensureFields()

1010 > s.compileSpec()

1011 >

1012 > s.updateCustomSearchAttributes(req.SearchAttributes)

1013 >

1014 > // Record customer start workflow memo payload size on each update.

1015 > s.recordActionPayloadMetrics()

1016 >

1017 > if s.hasMinVersion(UpdateFromPrevious) && !s.hasMinVersion(UseLastAction) {

1018 // We need to start re-processing from the last event, so that we catch actions whose

1019 // nominal time is before now but actual time (with jitter) is after now. Logic in

1025 }

1026

1027 > s.Info.UpdateTime = timestamppb.New(s.now()) workflow.go

1028 > s.incSeqNo()

1029 }

1030

1031 > func (s *scheduler) handleRefreshSignal(ch workflow.ReceiveChannel, _ bool) { workflow.go

1032 > ch.Receive(s.ctx, nil)

1033 > s.logger.Debug("got refresh signal")

1034 > // If we're woken up by any signal, we'll pass through ProcessBuffer before sleeping again.

1035 > // ProcessBuffer will see this flag and refresh everything.

1036 > s.State.NeedRefresh = true

1037 > }

1038

1039 func (s *scheduler) handleForceCANSignal(ch workflow.ReceiveChannel, _ bool) {

1086 }

1087

1088 > func (s *scheduler) processSignals() bool { workflow.go

1089 > scheduleChanged := false

1090 > if s.pendingPatch != nil {

1091 > s.processPatch(s.pendingPatch) workflow.go

1092 > s.pendingPatch = nil

1093 > }

1094 > if s.pendingUpdate != nil { workflow.go

1095 > s.processUpdate(s.pendingUpdate) workflow.go

1096 > s.pendingUpdate = nil

1097 > scheduleChanged = true

1098 > }

1099 > return scheduleChanged workflow.go

1100 }

1101

1105 // schedule's `RemainingActions` will be returned. Future action times that

1106 // precede the schedule's UpdateTime are not included.

1107 > func (s *scheduler) getFutureActionTimes(inWorkflowContext bool, n int) []*timestamppb.Timestamp { workflow.go

1108 > // Note that `s` may be a `scheduler` created outside of a workflow context, used to

1109 > // compute list info at creation time or in a query. In that case inWorkflowContext will

1110 > // be false, and this function and anything it calls should not use s.ctx.

1111 >

1112 > base := timestamp.TimeValue(s.State.LastProcessedTime)

1113 >

1114 > // Pure version not using workflow context

1115 > next := func(t time.Time) time.Time {

1116 > res, _ := s.cspec.GetNextTime(s.jitterSeed(), t) workflow.go

1117 > return res.Next

1118 > }

1119

1120 > if inWorkflowContext && s.hasMinVersion(NewCacheAndJitter) { workflow.go

1121 > // We can use the cache here

1122 > next = func(t time.Time) time.Time {

1123 > return s.getNextTimeV2(base, t).Next workflow.go

1124 > }

1125 }

1126

1127 > if s.hasMinVersion(AccurateFutureActionTimes) && s.Schedule.State.LimitedActions { workflow.go

1128 n = min(int(s.Schedule.State.RemainingActions), n)

1129 }

1130

1131 > if s.cspec == nil { workflow.go

1132 return nil

1133 }

1134 > out := make([]*timestamppb.Timestamp, 0, n) workflow.go

1135 > t1 := base

1136 > for len(out) < n {

1137 > t1 = next(t1)

1138 > if t1.IsZero() {

1139 > break workflow.go

1140 }

1141

1142 > if s.hasMinVersion(AccurateFutureActionTimes) && s.Info.UpdateTime.AsTime().After(t1) { workflow.go

1143 // Skip action times whose nominal times are prior to the schedule's update time

1144 continue

1145 }

1146

1147 > out = append(out, timestamppb.New(t1)) workflow.go

1148 }

1149 > return out workflow.go

1150 }

1151

1190 }

1191

1192 > func (s *scheduler) incSeqNo() { workflow.go

1193 > s.State.ConflictToken++

1194 > }

1195

1196 > func (s *scheduler) getListInfo(inWorkflowContext bool) *schedulepb.ScheduleListInfo { workflow.go

1197 > // Note that `s` may be a `scheduler` created outside of a workflow context, used to

1198 > // compute list info at creation time. In that case inWorkflowContext will be false,

1199 > // and this function and anything it calls should not use s.ctx.

1200 >

1201 > spec := common.CloneProto(s.Schedule.Spec)

1202 > // clear fields that are too large/not useful for the list view

1203 > spec.TimezoneData = nil

1204 >

1205 > if s.hasMinVersion(LimitMemoSpecSize) {

1206 > // Limit the number of specs and exclusions stored on the memo.

1207 > limit := s.tweakables.SpecFieldLengthLimit

1208 > spec.ExcludeStructuredCalendar = util.SliceHead(spec.ExcludeStructuredCalendar, limit)

1209 > spec.Interval = util.SliceHead(spec.Interval, limit)

1210 > spec.StructuredCalendar = util.SliceHead(spec.StructuredCalendar, limit)

1211 > }

1212

1213 > return &schedulepb.ScheduleListInfo{ workflow.go

1214 > Spec: spec,

1215 > WorkflowType: s.Schedule.Action.GetStartWorkflow().GetWorkflowType(),

1216 > Notes: s.Schedule.State.Notes,

1217 > Paused: s.Schedule.State.Paused,

1218 > RecentActions: util.SliceTail(s.Info.RecentActions, s.tweakables.RecentActionCountForList),

1219 > FutureActionTimes: s.getFutureActionTimes(inWorkflowContext, s.tweakables.FutureActionCountForList),

1220 > }

1221 }

1222

1223 > func (s *scheduler) updateCustomSearchAttributes(searchAttributes *commonpb.SearchAttributes) { workflow.go

1224 > // We want to distinguish nil value from an empty object value.

1225 > // When it is nil, then it's a no-op. Otherwise, it will overwrite the search attributes.

1226 > // That is, if it is an empty map, it will unset all search attributes.

1227 > if searchAttributes == nil {

1228 > return

1229 > }

1230

1231 upsertMap := map[string]any{}

1268 }

1269

1270 > func (s *scheduler) updateMemoAndSearchAttributes() { workflow.go

1271 > newInfo := s.getListInfo(true)

1272 >

1273 > workflowInfo := workflow.GetInfo(s.ctx)

1274 > currentInfoPayload := workflowInfo.Memo.GetFields()[MemoFieldInfo]

1275 >

1276 > var currentInfoBytes []byte

1277 > var currentInfo schedulepb.ScheduleListInfo

1278 >

1279 > //workflowcheck:ignore (proto.Equal is falsely flagged as non-deterministic)

1280 > if currentInfoPayload == nil ||

1281 > payload.Decode(currentInfoPayload, &currentInfoBytes) != nil ||

1282 > currentInfo.Unmarshal(currentInfoBytes) != nil ||

1283 > !proto.Equal(&currentInfo, newInfo) {

1284 > // marshal manually to get proto encoding (default dataconverter will use json)

1285 > newInfoBytes, err := newInfo.Marshal()

1286 > if err == nil {

1287 > err = workflow.UpsertMemo(s.ctx, map[string]any{

1288 > MemoFieldInfo: newInfoBytes,

1289 > })

1290 > }

1291 > if err != nil {

1292 s.logger.Error("error updating memo", "error", err)

1293 }

1295

1296 //nolint:staticcheck // SA1019: workflowInfo.SearchAttributes is not typed.

1297 > currentPausedPayload := workflowInfo.SearchAttributes.GetIndexedFields()[sadefs.TemporalSchedulePaused] workflow.go

1298 > var currentPaused bool

1299 > if currentPausedPayload == nil ||

1300 > payload.Decode(currentPausedPayload, &currentPaused) != nil ||

1301 > currentPaused != s.Schedule.State.Paused {

1302 > err := workflow.UpsertSearchAttributes(s.ctx, map[string]any{ //nolint:staticcheck // SA1019: untyped search attributes required here

1303 > sadefs.TemporalSchedulePaused: s.Schedule.State.Paused,

1304 > })

1305 > if err != nil {

1306 s.logger.Error("error updating search attributes", "error", err)

1307 }

1309 }

1310

1311 > func (s *scheduler) checkConflict(token int64) error { workflow.go

1312 > if token == 0 || token == s.State.ConflictToken {

1313 > return nil

1314 > }

1315 return errUpdateConflict

1316 }

1317

1318 > func (s *scheduler) updateTweakables() { workflow.go

1319 > // Use MutableSideEffect so that we can change the defaults without breaking determinism.

1320 > get := func(ctx workflow.Context) any {

1321 p := CurrentTweakablePolicies

1322 // Re-evaluates migration dynamic config each iteration.

1325 return p

1326 }

1327 > eq := func(a, b any) bool { return a.(TweakablePolicies) == b.(TweakablePolicies) } workflow.go

1328 > if err := workflow.MutableSideEffect(s.ctx, "tweakables", get, eq).Get(&s.tweakables); err != nil {

1329 panic("can't decode TweakablePolicies:" + err.Error())

1330 }

1331 }

1332

1333 > func (s *scheduler) getCatchupWindow() time.Duration { workflow.go

1334 > cw := s.Schedule.Policies.CatchupWindow

1335 > if cw == nil {

1336 > return s.tweakables.DefaultCatchupWindow workflow.go

1337 > } else if cw.AsDuration() < s.tweakables.MinCatchupWindow { workflow.go

1338 > return s.tweakables.MinCatchupWindow workflow.go

1339 > } else { workflow.go

1340 > return cw.AsDuration()

1341 > }

1342 }

1343

1344 > func (s *scheduler) resolveOverlapPolicy(overlapPolicy enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy { workflow.go

1345 > if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {

1346 > overlapPolicy = s.Schedule.Policies.OverlapPolicy

1347 > }

1348 > if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED {

1349 > overlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP workflow.go

1350 > }

1351 > return overlapPolicy workflow.go

1352 }

1353

1354 > func (s *scheduler) addStart(nominalTime, actualTime time.Time, overlapPolicy enumspb.ScheduleOverlapPolicy, manual bool) { workflow.go

1355 > s.logger.Debug("addStart", "start-time", nominalTime, "actual-start-time", actualTime, "overlap-policy", overlapPolicy, "manual", manual)

1356 > if s.tweakables.MaxBufferSize > 0 && len(s.State.BufferedStarts) >= s.tweakables.MaxBufferSize {

1357 s.logger.Warn("Buffer too large", "start-time", nominalTime, "overlap-policy", overlapPolicy, "manual", manual)

1358 s.metrics.Counter(metrics.ScheduleBufferOverruns.Name()).Inc(1)

1360 return

1361 }

1362 > s.State.BufferedStarts = append(s.State.BufferedStarts, &schedulespb.BufferedStart{ workflow.go

1363 > NominalTime: timestamppb.New(nominalTime),

1364 > ActualTime: timestamppb.New(actualTime),

1365 > OverlapPolicy: overlapPolicy,

1366 > Manual: manual,

1367 > })

1368 > // we have a new start to process, so we need to make sure that we have up-to-date status

1369 > // on any workflows that we started.

1370 > s.State.NeedRefresh = true

1371 }

1372

1374 //

1375 //nolint:revive

1376 > func (s *scheduler) processBuffer() bool { workflow.go

1377 > s.logger.Debug("ProcessBuffer", "buffer", len(s.State.BufferedStarts), "running", len(s.Info.RunningWorkflows), "need-refresh", s.State.NeedRefresh)

1378 >

1379 > // TODO: consider doing this always and removing needRefresh? we only end up here without

1380 > // needRefresh in the case of update, or patch without an immediate run, so it's not much

1381 > // wasted work.

1382 > // TODO: on the other hand, we don't have to refresh if we have one workflow running and a

1383 > // long-poll watcher running, because we would have gotten woken up already.

1384 > if s.State.NeedRefresh {

1385 > s.refreshWorkflows(slices.Clone(s.Info.RunningWorkflows)) workflow.go

1386 > s.State.NeedRefresh = false

1387 > }

1388

1389 // Make sure we have something to start. If not, we can clear the buffer.

1390 > req := s.Schedule.Action.GetStartWorkflow() workflow.go

1391 > if req == nil || len(s.State.BufferedStarts) == 0 {

1392 > s.State.BufferedStarts = nil workflow.go

1393 > return false

1394 > }

1395

1396 > isRunning := len(s.Info.RunningWorkflows) > 0 workflow.go

1397 > tryAgain := false

1398 > action := ProcessBuffer(s.State.BufferedStarts, isRunning, s.resolveOverlapPolicy)

1399 >

1400 > s.State.BufferedStarts = action.NewBuffer

1401 > s.Info.OverlapSkipped += action.OverlapSkipped

1402 > for overlapPolicy, count := range action.OverlapSkippedByPolicy {

1403 > s.metrics.WithTags(map[string]string{ workflow.go

1404 > metrics.ScheduleOverlapPolicyTag: overlapPolicy.String(),

1405 > }).Counter(metrics.ScheduleOverlapSkipped.Name()).Inc(count)

1406 > }

1407

1408 // Try starting whatever we're supposed to start now

1409 > allStarts := action.OverlappingStarts workflow.go

1410 > if action.NonOverlappingStart != nil {

1411 > allStarts = append(allStarts, action.NonOverlappingStart) workflow.go

1412 > }

1413 > for _, start := range allStarts { workflow.go

1414 > if !s.canTakeScheduledAction(start.Manual, true) {

1415 > // try again to drain the buffer if paused or out of actions workflow.go

1416 > tryAgain = true

1417 > continue

1418 }

1419 > result, err := s.startWorkflow(start, req) workflow.go

1420 > metricsWithTag := s.metrics.WithTags(map[string]string{

1421 > metrics.ScheduleActionTypeTag: metrics.ScheduleActionStartWorkflow,

1422 > })

1423 > if err != nil {

1424 s.logger.Error("Failed to start workflow", "error", err)

1425 if !isUserScheduleError(err) {

1429 continue

1430 }

1431 > metricsWithTag.Counter(metrics.ScheduleActionSuccess.Name()).Inc(1) workflow.go

1432 > nonOverlapping := start == action.NonOverlappingStart

1433 > s.recordAction(result, nonOverlapping)

1434 }

1435

1436 // Terminate or cancel if required (terminate overrides cancel if both are present)

1437 > if action.NeedTerminate { workflow.go

1438 for _, ex := range s.Info.RunningWorkflows {

1439 s.terminateWorkflow(ex)

1440 }

1441 > } else if action.NeedCancel { workflow.go

1442 for _, ex := range s.Info.RunningWorkflows {

1443 s.cancelWorkflow(ex)

1449 // one of them with an activity. We only need one watcher at a time, though: after that one

1450 // returns, we'll end up back here and start the next one.

1451 > if len(s.State.BufferedStarts) > 0 && s.watchingFuture == nil { workflow.go

1452 > if len(s.Info.RunningWorkflows) > 0 { workflow.go

1453 > s.startLongPollWatcher(s.Info.RunningWorkflows[0])

1454 > } else {

1455 > s.logger.Error("have buffered workflows but none running") workflow.go

1456 > }

1457 }

1458

1459 > return tryAgain workflow.go

1460 }

1461

1462 > func (s *scheduler) recordAction(result *schedulepb.ScheduleActionResult, nonOverlapping bool) { workflow.go

1463 > s.Info.ActionCount++

1464 > s.Info.RecentActions = util.SliceTail(append(s.Info.RecentActions, result), s.tweakables.RecentActionCount)

1465 > canTrack := nonOverlapping || !s.hasMinVersion(DontTrackOverlapping)

1466 > if canTrack && result.StartWorkflowResult != nil {

1467 > s.Info.RunningWorkflows = append(s.Info.RunningWorkflows, result.StartWorkflowResult) workflow.go

1468 > }

1469 }

1470

1472 start *schedulespb.BufferedStart,

1473 newWorkflow *workflowpb.NewWorkflowExecutionInfo,

1474 > ) (*schedulepb.ScheduleActionResult, error) { workflow.go

1475 > nominalTimeSec := start.NominalTime.AsTime().UTC().Truncate(time.Second)

1476 > workflowID := newWorkflow.WorkflowId

1477 > if start.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL || s.tweakables.AlwaysAppendTimestamp {

1478 > // must match AppendedTimestampForValidation

1479 > workflowID += "-" + nominalTimeSec.Format(time.RFC3339)

1480 > }

1481

1482 // Set scheduleToCloseTimeout based on catchup window, which is the latest time that it's

1483 // acceptable to start this workflow. For manual starts (trigger immediately or backfill),

1484 // catch up window doesn't apply, so just use 60s.

1485 > options := defaultLocalActivityOptions() workflow.go

1486 > if start.Manual {

1487 > options.ScheduleToCloseTimeout = 60 * time.Second workflow.go

1488 > } else { workflow.go

1489 > deadline := start.ActualTime.AsTime().Add(s.getCatchupWindow()) workflow.go

1490 > options.ScheduleToCloseTimeout = deadline.Sub(s.now())

1491 > if options.ScheduleToCloseTimeout < options.StartToCloseTimeout {

1492 options.ScheduleToCloseTimeout = options.StartToCloseTimeout

1493 > } else if options.ScheduleToCloseTimeout > 1*time.Hour { workflow.go

1494 > options.ScheduleToCloseTimeout = 1 * time.Hour workflow.go

1495 > }

1496 }

1497 > ctx := workflow.WithLocalActivityOptions(s.ctx, options) workflow.go

1498 >

1499 > lastCompletionResult, continuedFailure := s.State.LastCompletionResult, s.State.ContinuedFailure

1500 > if start.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL && s.hasMinVersion(DontTrackOverlapping) {

1501 > // ALLOW_ALL runs don't participate in lastCompletionResult/continuedFailure at all workflow.go

1502 > lastCompletionResult = nil

1503 > continuedFailure = nil

1504 > }

1505

1506 // Reject duplicates as part of WFID reuse policy when possible, as a measure

1507 // against WFT timeouts/failures that lead to non-determinism.

1508 > reusePolicy := enumspb.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE workflow.go

1509 > if start.Manual {

1510 > reusePolicy = enumspb.WORKFLOW_ID_REUSE_POLICY_ALLOW_DUPLICATE workflow.go

1511 > }

1512

1513 > req := &schedulespb.StartWorkflowRequest{ workflow.go

1514 > Request: &workflowservice.StartWorkflowExecutionRequest{

1515 > WorkflowId: workflowID,

1516 > WorkflowType: newWorkflow.WorkflowType,

1517 > TaskQueue: newWorkflow.TaskQueue,

1518 > Input: newWorkflow.Input,

1519 > WorkflowExecutionTimeout: newWorkflow.WorkflowExecutionTimeout,

1520 > WorkflowRunTimeout: newWorkflow.WorkflowRunTimeout,

1521 > WorkflowTaskTimeout: newWorkflow.WorkflowTaskTimeout,

1522 > Identity: s.identity(),

1523 > RequestId: s.newUUIDString(),

1524 > WorkflowIdReusePolicy: reusePolicy,

1525 > RetryPolicy: newWorkflow.RetryPolicy,

1526 > Memo: newWorkflow.Memo,

1527 > SearchAttributes: s.addSearchAttributes(newWorkflow.SearchAttributes, nominalTimeSec),

1528 > Header: newWorkflow.Header,

1529 > LastCompletionResult: lastCompletionResult,

1530 > ContinuedFailure: continuedFailure,

1531 > UserMetadata: newWorkflow.UserMetadata,

1532 > Priority: newWorkflow.Priority,

1533 > },

1534 > }

1535 > for {

1536 > var res schedulespb.StartWorkflowResponse

1537 > err := workflow.ExecuteLocalActivity(ctx, s.a.StartWorkflow, req).Get(s.ctx, &res)

1538 > var appErr *temporal.ApplicationError

1539 > var details rateLimitedDetails

1540 > if errors.As(err, &appErr) && appErr.Type() == rateLimitedErrorType && appErr.Details(&details) == nil {

1541 s.metrics.Counter(metrics.ScheduleRateLimited.Name()).Inc(1)

1542 workflow.Sleep(s.ctx, details.Delay)

1544 continue

1545 }

1546 > if err != nil { workflow.go

1547 return nil, err

1548 }

1549

1550 > if !start.Manual { workflow.go

1551 > // record metric only for _scheduled_ actions, not trigger/backfill, otherwise it's not meaningful workflow.go

1552 > desiredTime := cmp.Or(start.DesiredTime, start.ActualTime)

1553 > s.metrics.Timer(metrics.ScheduleActionDelay.Name()).Record(res.RealStartTime.AsTime().Sub(desiredTime.AsTime()))

1554 > // Record total delay from original schedule time, including any overlap policy wait.

1555 > s.metrics.Timer(metrics.ScheduleActionE2EDelay.Name()).Record(res.RealStartTime.AsTime().Sub(start.ActualTime.AsTime()))

1556 > }

1557

1558 > actionResult := &schedulepb.ScheduleActionResult{ workflow.go

1559 > ScheduleTime: start.ActualTime,

1560 > ActualTime: res.RealStartTime,

1561 > StartWorkflowResult: &commonpb.WorkflowExecution{

1562 > WorkflowId: workflowID,

1563 > RunId: res.RunId,

1564 > },

1565 > }

1566 >

1567 > if s.hasMinVersion(ActionResultIncludesStatus) {

1568 actionResult.StartWorkflowStatus = enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING

1569 }

1570

1571 > return actionResult, nil workflow.go

1572 }

1573 }

1574

1575 > func (s *scheduler) identity() string { workflow.go

1576 > return fmt.Sprintf("temporal-scheduler-%s-%s", s.State.Namespace, s.State.ScheduleId)

1577 > }

1578

1579 > func (s *scheduler) jitterSeed() string { workflow.go

1580 > if s.hasMinVersion(NewCacheAndJitter) {

1581 return fmt.Sprintf("%s-%s", s.State.NamespaceId, s.State.ScheduleId)

1582 }

1583 > return "" workflow.go

1584 }

1585

1587 attributes *commonpb.SearchAttributes,

1588 nominal time.Time,

1589 > ) *commonpb.SearchAttributes { workflow.go

1590 > fields := util.CloneMapNonNil(attributes.GetIndexedFields())

1591 > if p, err := payload.Encode(nominal); err == nil {

1592 > fields[sadefs.TemporalScheduledStartTime] = p

1593 > }

1594 > if p, err := payload.Encode(s.State.ScheduleId); err == nil {

1595 > fields[sadefs.TemporalScheduledById] = p

1596 > }

1597 > return &commonpb.SearchAttributes{

1598 > IndexedFields: fields,

1599 > }

1600 }

1601

1602 > func (s *scheduler) refreshWorkflows(executions []*commonpb.WorkflowExecution) { workflow.go

1603 > ctx := workflow.WithLocalActivityOptions(s.ctx, defaultLocalActivityOptions())

1604 > futures := make([]workflow.Future, len(executions))

1605 > for i, ex := range executions {

1606 > req := &schedulespb.WatchWorkflowRequest{ workflow.go

1607 > // Note: do not send runid here so that we always get the latest one

1608 > Execution: &commonpb.WorkflowExecution{WorkflowId: ex.WorkflowId},

1609 > FirstExecutionRunId: ex.RunId,

1610 > LongPoll: false,

1611 > }

1612 > futures[i] = workflow.ExecuteLocalActivity(ctx, s.a.WatchWorkflow, req)

1613 > }

1614 > for i, ex := range executions { workflow.go

1615 > s.processWatcherResult(ex.WorkflowId, futures[i], false) workflow.go

1616 > }

1617 }

1618

1619 > func (s *scheduler) startLongPollWatcher(ex *commonpb.WorkflowExecution) { workflow.go

1620 > if s.watchingFuture != nil {

1621 s.logger.Error("startLongPollWatcher called with watcher already running")

1622 return

1623 }

1624

1625 > ctx := workflow.WithActivityOptions(s.ctx, workflow.ActivityOptions{ workflow.go

1626 > StartToCloseTimeout: 365 * 24 * time.Hour,

1627 > RetryPolicy: &temporal.RetryPolicy{

1628 > InitialInterval: 1 * time.Second,

1629 > MaximumInterval: 30 * time.Second,

1630 > },

1631 > HeartbeatTimeout: 65 * time.Second,

1632 > })

1633 > req := &schedulespb.WatchWorkflowRequest{

1634 > // Note: do not send runid here so that we always get the latest one

1635 > Execution: &commonpb.WorkflowExecution{WorkflowId: ex.WorkflowId},

1636 > FirstExecutionRunId: ex.RunId,

1637 > LongPoll: true,

1638 > }

1639 > s.watchingFuture = workflow.ExecuteActivity(ctx, s.a.WatchWorkflow, req)

1640 > s.watchingWorkflowId = ex.WorkflowId

1641 }

1642

1677 }

1678

1679 > func (s *scheduler) getRetentionExpiration(nextWakeup time.Time) time.Time { workflow.go

1680 > // if RetentionTime is not set or the schedule is paused or nextWakeup time is not zero

1681 > // or there is more action to take, there is no need for retention

1682 > if s.tweakables.RetentionTime == 0 ||

1683 > s.Schedule.State.Paused ||

1684 > (!nextWakeup.IsZero() && s.canTakeScheduledAction(false, false)) ||

1685 > s.hasMoreAllowAllBackfills() {

1686 > return time.Time{} workflow.go

1687 > }

1688

1689 // retention starts from the last "event"

1690 > return s.getLastEvent().Add(s.tweakables.RetentionTime) workflow.go

1691 }

1692

1695 // retention time (how long an idle schedule lives after becoming idle), and also for

1696 // recalculating times after an update to account for jitter.

1697 > func (s *scheduler) getLastEvent() time.Time { workflow.go

1698 > var lastEvent time.Time

1699 > if len(s.Info.RecentActions) > 0 {

1700 > lastEvent = s.Info.RecentActions[len(s.Info.RecentActions)-1].ActualTime.AsTime() workflow.go

1701 > }

1702 > lastEvent = util.MaxTime(lastEvent, s.Info.CreateTime.AsTime()) workflow.go

1703 > lastEvent = util.MaxTime(lastEvent, s.Info.UpdateTime.AsTime())

1704 > return lastEvent

1705 }

1706

1707 > func (s *scheduler) newUUIDString() string { workflow.go

1708 > if len(s.uuidBatch) == 0 {

1709 > panicIfErr(workflow.SideEffect(s.ctx, func(ctx workflow.Context) any {

1710 out := make([]string, 10)

1711 for i := range out {

1715 }).Get(&s.uuidBatch))

1716 }

1717 > next := s.uuidBatch[0] workflow.go

1718 > s.uuidBatch = s.uuidBatch[1:]

1719 > return next

1720 }

1721

1722 > func (s *scheduler) hasMinVersion(version SchedulerWorkflowVersion) bool { workflow.go

1723 > return s.tweakables.Version >= version

1724 > }

1725

1726 > func panicIfErr(err error) { workflow.go

1727 > if err != nil {

1728 panic(err)

1729 }

go.temporal.io/server/common/dynamicconfig/setting_gen.go 245 covered LOC · 51 ranges

Open complete file

26 type GlobalBoolConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[bool]

27

28 > func NewGlobalBoolSetting(key string, def bool, description string) GlobalBoolSetting { setting_gen.go

29 > return NewGlobalTypedSettingWithConverter[bool](key, convertBool, def, description)

30 > }

31

32 func NewGlobalBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) GlobalBoolConstrainedDefaultSetting {

43 type NamespaceBoolConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[bool]

44

45 > func NewNamespaceBoolSetting(key string, def bool, description string) NamespaceBoolSetting { setting_gen.go

46 > return NewNamespaceTypedSettingWithConverter[bool](key, convertBool, def, description)

47 > }

48

49 func NewNamespaceBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) NamespaceBoolConstrainedDefaultSetting {

60 type NamespaceIDBoolConstrainedDefaultSetting = NamespaceIDTypedConstrainedDefaultSetting[bool]

61

62 > func NewNamespaceIDBoolSetting(key string, def bool, description string) NamespaceIDBoolSetting { setting_gen.go

63 > return NewNamespaceIDTypedSettingWithConverter[bool](key, convertBool, def, description)

64 > }

65

66 func NewNamespaceIDBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) NamespaceIDBoolConstrainedDefaultSetting {

77 type TaskQueueBoolConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[bool]

78

79 > func NewTaskQueueBoolSetting(key string, def bool, description string) TaskQueueBoolSetting { setting_gen.go

80 > return NewTaskQueueTypedSettingWithConverter[bool](key, convertBool, def, description)

81 > }

82

83 func NewTaskQueueBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) TaskQueueBoolConstrainedDefaultSetting {

128 type DestinationBoolConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[bool]

129

130 > func NewDestinationBoolSetting(key string, def bool, description string) DestinationBoolSetting { setting_gen.go

131 > return NewDestinationTypedSettingWithConverter[bool](key, convertBool, def, description)

132 > }

133

134 func NewDestinationBoolSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[bool], description string) DestinationBoolConstrainedDefaultSetting {

162 type GlobalIntConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[int]

163

164 > func NewGlobalIntSetting(key string, def int, description string) GlobalIntSetting { setting_gen.go

165 > return NewGlobalTypedSettingWithConverter[int](key, convertInt, def, description)

166 > }

167

168 func NewGlobalIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) GlobalIntConstrainedDefaultSetting {

179 type NamespaceIntConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[int]

180

181 > func NewNamespaceIntSetting(key string, def int, description string) NamespaceIntSetting { setting_gen.go

182 > return NewNamespaceTypedSettingWithConverter[int](key, convertInt, def, description)

183 > }

184

185 func NewNamespaceIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) NamespaceIntConstrainedDefaultSetting {

189 type IntPropertyFnWithNamespaceFilter = TypedPropertyFnWithNamespaceFilter[int]

190

191 > func GetIntPropertyFnFilteredByNamespace(value int) IntPropertyFnWithNamespaceFilter { setting_gen.go

192 > return GetTypedPropertyFnFilteredByNamespace(value)

193 > }

194

195 type NamespaceIDIntSetting = NamespaceIDTypedSetting[int]

213 type TaskQueueIntConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[int]

214

215 > func NewTaskQueueIntSetting(key string, def int, description string) TaskQueueIntSetting { setting_gen.go

216 > return NewTaskQueueTypedSettingWithConverter[int](key, convertInt, def, description)

217 > }

218

219 > func NewTaskQueueIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) TaskQueueIntConstrainedDefaultSetting { setting_gen.go

220 > return NewTaskQueueTypedSettingWithConstrainedDefault[int](key, convertInt, cdef, description)

221 > }

222

223 type IntPropertyFnWithTaskQueueFilter = TypedPropertyFnWithTaskQueueFilter[int]

230 type ShardIDIntConstrainedDefaultSetting = ShardIDTypedConstrainedDefaultSetting[int]

231

232 > func NewShardIDIntSetting(key string, def int, description string) ShardIDIntSetting { setting_gen.go

233 > return NewShardIDTypedSettingWithConverter[int](key, convertInt, def, description)

234 > }

235

236 func NewShardIDIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) ShardIDIntConstrainedDefaultSetting {

264 type DestinationIntConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[int]

265

266 > func NewDestinationIntSetting(key string, def int, description string) DestinationIntSetting { setting_gen.go

267 > return NewDestinationTypedSettingWithConverter[int](key, convertInt, def, description)

268 > }

269

270 func NewDestinationIntSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[int], description string) DestinationIntConstrainedDefaultSetting {

298 type GlobalFloatConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[float64]

299

300 > func NewGlobalFloatSetting(key string, def float64, description string) GlobalFloatSetting { setting_gen.go

301 > return NewGlobalTypedSettingWithConverter[float64](key, convertFloat, def, description)

302 > }

303

304 func NewGlobalFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) GlobalFloatConstrainedDefaultSetting {

315 type NamespaceFloatConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[float64]

316

317 > func NewNamespaceFloatSetting(key string, def float64, description string) NamespaceFloatSetting { setting_gen.go

318 > return NewNamespaceTypedSettingWithConverter[float64](key, convertFloat, def, description)

319 > }

320

321 func NewNamespaceFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) NamespaceFloatConstrainedDefaultSetting {

349 type TaskQueueFloatConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[float64]

350

351 > func NewTaskQueueFloatSetting(key string, def float64, description string) TaskQueueFloatSetting { setting_gen.go

352 > return NewTaskQueueTypedSettingWithConverter[float64](key, convertFloat, def, description)

353 > }

354

355 func NewTaskQueueFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) TaskQueueFloatConstrainedDefaultSetting {

366 type ShardIDFloatConstrainedDefaultSetting = ShardIDTypedConstrainedDefaultSetting[float64]

367

368 > func NewShardIDFloatSetting(key string, def float64, description string) ShardIDFloatSetting { setting_gen.go

369 > return NewShardIDTypedSettingWithConverter[float64](key, convertFloat, def, description)

370 > }

371

372 func NewShardIDFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) ShardIDFloatConstrainedDefaultSetting {

400 type DestinationFloatConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[float64]

401

402 > func NewDestinationFloatSetting(key string, def float64, description string) DestinationFloatSetting { setting_gen.go

403 > return NewDestinationTypedSettingWithConverter[float64](key, convertFloat, def, description)

404 > }

405

406 func NewDestinationFloatSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[float64], description string) DestinationFloatConstrainedDefaultSetting {

434 type GlobalStringConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[string]

435

436 > func NewGlobalStringSetting(key string, def string, description string) GlobalStringSetting { setting_gen.go

437 > return NewGlobalTypedSettingWithConverter[string](key, convertString, def, description)

438 > }

439

440 func NewGlobalStringSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[string], description string) GlobalStringConstrainedDefaultSetting {

570 type GlobalDurationConstrainedDefaultSetting = GlobalTypedConstrainedDefaultSetting[time.Duration]

571

572 > func NewGlobalDurationSetting(key string, def time.Duration, description string) GlobalDurationSetting { setting_gen.go

573 > return NewGlobalTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

574 > }

575

576 func NewGlobalDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) GlobalDurationConstrainedDefaultSetting {

587 type NamespaceDurationConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[time.Duration]

588

589 > func NewNamespaceDurationSetting(key string, def time.Duration, description string) NamespaceDurationSetting { setting_gen.go

590 > return NewNamespaceTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

591 > }

592

593 func NewNamespaceDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) NamespaceDurationConstrainedDefaultSetting {

604 type NamespaceIDDurationConstrainedDefaultSetting = NamespaceIDTypedConstrainedDefaultSetting[time.Duration]

605

606 > func NewNamespaceIDDurationSetting(key string, def time.Duration, description string) NamespaceIDDurationSetting { setting_gen.go

607 > return NewNamespaceIDTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

608 > }

609

610 func NewNamespaceIDDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) NamespaceIDDurationConstrainedDefaultSetting {

621 type TaskQueueDurationConstrainedDefaultSetting = TaskQueueTypedConstrainedDefaultSetting[time.Duration]

622

623 > func NewTaskQueueDurationSetting(key string, def time.Duration, description string) TaskQueueDurationSetting { setting_gen.go

624 > return NewTaskQueueTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

625 > }

626

627 > func NewTaskQueueDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) TaskQueueDurationConstrainedDefaultSetting { setting_gen.go

628 > return NewTaskQueueTypedSettingWithConstrainedDefault[time.Duration](key, convertDuration, cdef, description)

629 > }

630

631 type DurationPropertyFnWithTaskQueueFilter = TypedPropertyFnWithTaskQueueFilter[time.Duration]

638 type ShardIDDurationConstrainedDefaultSetting = ShardIDTypedConstrainedDefaultSetting[time.Duration]

639

640 > func NewShardIDDurationSetting(key string, def time.Duration, description string) ShardIDDurationSetting { setting_gen.go

641 > return NewShardIDTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

642 > }

643

644 func NewShardIDDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) ShardIDDurationConstrainedDefaultSetting {

655 type TaskTypeDurationConstrainedDefaultSetting = TaskTypeTypedConstrainedDefaultSetting[time.Duration]

656

657 > func NewTaskTypeDurationSetting(key string, def time.Duration, description string) TaskTypeDurationSetting { setting_gen.go

658 > return NewTaskTypeTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

659 > }

660

661 func NewTaskTypeDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) TaskTypeDurationConstrainedDefaultSetting {

672 type DestinationDurationConstrainedDefaultSetting = DestinationTypedConstrainedDefaultSetting[time.Duration]

673

674 > func NewDestinationDurationSetting(key string, def time.Duration, description string) DestinationDurationSetting { setting_gen.go

675 > return NewDestinationTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

676 > }

677

678 func NewDestinationDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) DestinationDurationConstrainedDefaultSetting {

689 type ChasmTaskTypeDurationConstrainedDefaultSetting = ChasmTaskTypeTypedConstrainedDefaultSetting[time.Duration]

690

691 > func NewChasmTaskTypeDurationSetting(key string, def time.Duration, description string) ChasmTaskTypeDurationSetting { setting_gen.go

692 > return NewChasmTaskTypeTypedSettingWithConverter[time.Duration](key, convertDuration, def, description)

693 > }

694

695 func NewChasmTaskTypeDurationSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[time.Duration], description string) ChasmTaskTypeDurationConstrainedDefaultSetting {

723 type NamespaceMapConstrainedDefaultSetting = NamespaceTypedConstrainedDefaultSetting[map[string]any]

724

725 > func NewNamespaceMapSetting(key string, def map[string]any, description string) NamespaceMapSetting { setting_gen.go

726 > return NewNamespaceTypedSettingWithConverter[map[string]any](key, convertMap, def, description)

727 > }

728

729 func NewNamespaceMapSettingWithConstrainedDefault(key string, cdef []TypedConstrainedValue[map[string]any], description string) NamespaceMapConstrainedDefaultSetting {

845 // values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful

846 // when using non-empty maps or slices as defaults, the result may not be what you want.

847 > func NewGlobalTypedSetting[T any](key string, def T, description string) GlobalTypedSetting[T] { setting_gen.go

848 > // Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.

849 > warnDefaultSharedStructure(key, def)

850 > // If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.

851 > _ = deepCopyForMapstructure(def)

852 >

853 > s := GlobalTypedSetting[T]{

854 > key: MakeKey(key),

855 > def: def,

856 > convert: ConvertStructure[T](def),

857 > description: description,

858 > }

859 > register(s)

860 > return s

861 > }

862

863 // NewGlobalTypedSettingWithConverter creates a setting with a custom converter function.

864 > func NewGlobalTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) GlobalTypedSetting[T] { setting_gen.go

865 > s := GlobalTypedSetting[T]{

866 > key: MakeKey(key),

867 > def: def,

868 > convert: convert,

869 > description: description,

870 > }

871 > register(s)

872 > return s

873 > }

874

875 // NewGlobalTypedSettingWithConstrainedDefault creates a setting with a compound default value.

885 }

886

887 > func (s GlobalTypedSetting[T]) Key() Key { return s.key } setting_gen.go

888 func (s GlobalTypedSetting[T]) Precedence() Precedence { return PrecedenceGlobal }

889 func (s GlobalTypedSetting[T]) Validate(v any) error {

981 // values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful

982 // when using non-empty maps or slices as defaults, the result may not be what you want.

983 > func NewNamespaceTypedSetting[T any](key string, def T, description string) NamespaceTypedSetting[T] { setting_gen.go

984 > // Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.

985 > warnDefaultSharedStructure(key, def)

986 > // If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.

987 > _ = deepCopyForMapstructure(def)

988 >

989 > s := NamespaceTypedSetting[T]{

990 > key: MakeKey(key),

991 > def: def,

992 > convert: ConvertStructure[T](def),

993 > description: description,

994 > }

995 > register(s)

996 > return s

997 > }

998

999 // NewNamespaceTypedSettingWithConverter creates a setting with a custom converter function.

1000 > func NewNamespaceTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) NamespaceTypedSetting[T] { setting_gen.go

1001 > s := NamespaceTypedSetting[T]{

1002 > key: MakeKey(key),

1003 > def: def,

1004 > convert: convert,

1005 > description: description,

1006 > }

1007 > register(s)

1008 > return s

1009 > }

1010

1011 // NewNamespaceTypedSettingWithConstrainedDefault creates a setting with a compound default value.

1021 }

1022

1023 > func (s NamespaceTypedSetting[T]) Key() Key { return s.key } setting_gen.go

1024 func (s NamespaceTypedSetting[T]) Precedence() Precedence { return PrecedenceNamespace }

1025 func (s NamespaceTypedSetting[T]) Validate(v any) error {

1105 }

1106

1107 > func GetTypedPropertyFnFilteredByNamespace[T any](value T) TypedPropertyFnWithNamespaceFilter[T] { setting_gen.go

1108 > return func(namespace string) T {

1109 return value

1110 }

1134

1135 // NewNamespaceIDTypedSettingWithConverter creates a setting with a custom converter function.

1136 > func NewNamespaceIDTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) NamespaceIDTypedSetting[T] { setting_gen.go

1137 > s := NamespaceIDTypedSetting[T]{

1138 > key: MakeKey(key),

1139 > def: def,

1140 > convert: convert,

1141 > description: description,

1142 > }

1143 > register(s)

1144 > return s

1145 > }

1146

1147 // NewNamespaceIDTypedSettingWithConstrainedDefault creates a setting with a compound default value.

1157 }

1158

1159 > func (s NamespaceIDTypedSetting[T]) Key() Key { return s.key } setting_gen.go

1160 func (s NamespaceIDTypedSetting[T]) Precedence() Precedence { return PrecedenceNamespaceID }

1161 func (s NamespaceIDTypedSetting[T]) Validate(v any) error {

1253 // values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful

1254 // when using non-empty maps or slices as defaults, the result may not be what you want.

1255 > func NewTaskQueueTypedSetting[T any](key string, def T, description string) TaskQueueTypedSetting[T] { setting_gen.go

1256 > // Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.

1257 > warnDefaultSharedStructure(key, def)

1258 > // If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.

1259 > _ = deepCopyForMapstructure(def)

1260 >

1261 > s := TaskQueueTypedSetting[T]{

1262 > key: MakeKey(key),

1263 > def: def,

1264 > convert: ConvertStructure[T](def),

1265 > description: description,

1266 > }

1267 > register(s)

1268 > return s

1269 > }

1270

1271 // NewTaskQueueTypedSettingWithConverter creates a setting with a custom converter function.

1272 > func NewTaskQueueTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) TaskQueueTypedSetting[T] { setting_gen.go

1273 > s := TaskQueueTypedSetting[T]{

1274 > key: MakeKey(key),

1275 > def: def,

1276 > convert: convert,

1277 > description: description,

1278 > }

1279 > register(s)

1280 > return s

1281 > }

1282

1283 // NewTaskQueueTypedSettingWithConstrainedDefault creates a setting with a compound default value.

1284 > func NewTaskQueueTypedSettingWithConstrainedDefault[T any](key string, convert func(any) (T, error), cdef []TypedConstrainedValue[T], description string) TaskQueueTypedConstrainedDefaultSetting[T] { setting_gen.go

1285 > s := TaskQueueTypedConstrainedDefaultSetting[T]{

1286 > key: MakeKey(key),

1287 > cdef: cdef,

1288 > convert: convert,

1289 > description: description,

1290 > }

1291 > register(s)

1292 > return s

1293 > }

1294

1295 > func (s TaskQueueTypedSetting[T]) Key() Key { return s.key } setting_gen.go

1296 func (s TaskQueueTypedSetting[T]) Precedence() Precedence { return PrecedenceTaskQueue }

1297 func (s TaskQueueTypedSetting[T]) Validate(v any) error {

1300 }

1301

1302 > func (s TaskQueueTypedConstrainedDefaultSetting[T]) Key() Key { return s.key } setting_gen.go

1303 func (s TaskQueueTypedConstrainedDefaultSetting[T]) Precedence() Precedence { return PrecedenceTaskQueue }

1304 func (s TaskQueueTypedConstrainedDefaultSetting[T]) Validate(v any) error {

1430

1431 // NewShardIDTypedSettingWithConverter creates a setting with a custom converter function.

1432 > func NewShardIDTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) ShardIDTypedSetting[T] { setting_gen.go

1433 > s := ShardIDTypedSetting[T]{

1434 > key: MakeKey(key),

1435 > def: def,

1436 > convert: convert,

1437 > description: description,

1438 > }

1439 > register(s)

1440 > return s

1441 > }

1442

1443 // NewShardIDTypedSettingWithConstrainedDefault creates a setting with a compound default value.

1453 }

1454

1455 > func (s ShardIDTypedSetting[T]) Key() Key { return s.key } setting_gen.go

1456 func (s ShardIDTypedSetting[T]) Precedence() Precedence { return PrecedenceShardID }

1457 func (s ShardIDTypedSetting[T]) Validate(v any) error {

1566

1567 // NewTaskTypeTypedSettingWithConverter creates a setting with a custom converter function.

1568 > func NewTaskTypeTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) TaskTypeTypedSetting[T] { setting_gen.go

1569 > s := TaskTypeTypedSetting[T]{

1570 > key: MakeKey(key),

1571 > def: def,

1572 > convert: convert,

1573 > description: description,

1574 > }

1575 > register(s)

1576 > return s

1577 > }

1578

1579 // NewTaskTypeTypedSettingWithConstrainedDefault creates a setting with a compound default value.

1589 }

1590

1591 > func (s TaskTypeTypedSetting[T]) Key() Key { return s.key } setting_gen.go

1592 func (s TaskTypeTypedSetting[T]) Precedence() Precedence { return PrecedenceTaskType }

1593 func (s TaskTypeTypedSetting[T]) Validate(v any) error {

1685 // values. The value from dynamic config will be _merged_ over a deep copy of 'def'. Be very careful

1686 // when using non-empty maps or slices as defaults, the result may not be what you want.

1687 > func NewDestinationTypedSetting[T any](key string, def T, description string) DestinationTypedSetting[T] { setting_gen.go

1688 > // Warn on any shared structure used with ConvertStructure, even though we handle it by deep copying.

1689 > warnDefaultSharedStructure(key, def)

1690 > // If even deep copy won't even work, we should panic early. Do that by calling deep copy once here.

1691 > _ = deepCopyForMapstructure(def)

1692 >

1693 > s := DestinationTypedSetting[T]{

1694 > key: MakeKey(key),

1695 > def: def,

1696 > convert: ConvertStructure[T](def),

1697 > description: description,

1698 > }

1699 > register(s)

1700 > return s

1701 > }

1702

1703 // NewDestinationTypedSettingWithConverter creates a setting with a custom converter function.

1704 > func NewDestinationTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) DestinationTypedSetting[T] { setting_gen.go

1705 > s := DestinationTypedSetting[T]{

1706 > key: MakeKey(key),

1707 > def: def,

1708 > convert: convert,

1709 > description: description,

1710 > }

1711 > register(s)

1712 > return s

1713 > }

1714

1715 // NewDestinationTypedSettingWithConstrainedDefault creates a setting with a compound default value.

1725 }

1726

1727 > func (s DestinationTypedSetting[T]) Key() Key { return s.key } setting_gen.go

1728 func (s DestinationTypedSetting[T]) Precedence() Precedence { return PrecedenceDestination }

1729 func (s DestinationTypedSetting[T]) Validate(v any) error {

1858

1859 // NewChasmTaskTypeTypedSettingWithConverter creates a setting with a custom converter function.

1860 > func NewChasmTaskTypeTypedSettingWithConverter[T any](key string, convert func(any) (T, error), def T, description string) ChasmTaskTypeTypedSetting[T] { setting_gen.go

1861 > s := ChasmTaskTypeTypedSetting[T]{

1862 > key: MakeKey(key),

1863 > def: def,

1864 > convert: convert,

1865 > description: description,

1866 > }

1867 > register(s)

1868 > return s

1869 > }

1870

1871 // NewChasmTaskTypeTypedSettingWithConstrainedDefault creates a setting with a compound default value.

1881 }

1882

1883 > func (s ChasmTaskTypeTypedSetting[T]) Key() Key { return s.key } setting_gen.go

1884 func (s ChasmTaskTypeTypedSetting[T]) Precedence() Precedence { return PrecedenceChasmTaskType }

1885 func (s ChasmTaskTypeTypedSetting[T]) Validate(v any) error {

go.temporal.io/server/service/worker/scheduler/spec.go 169 covered LOC · 55 ranges

Open complete file

63 // NewSpecBuilder takes the compute-limit getters directly (rather than a *dynamicconfig.Collection)

64 // so the dynamic-config plumbing stays in the wiring layer, per the common codebase pattern.

65 > func NewSpecBuilder(warnIterations, maxIterations dynamicconfig.IntPropertyFn) *SpecBuilder { spec.go

66 > return &SpecBuilder{

67 > warnIterations: warnIterations,

68 > maxIterations: maxIterations,

69 > locationCache: cache.New(1000,

70 > &cache.Options{

71 > TTL: 24 * time.Hour,

72 > },

73 > ),

74 > }

75 > }

76

77 > func (b *SpecBuilder) NewCompiledSpec(spec *schedulepb.ScheduleSpec) (*CompiledSpec, error) { spec.go

78 > spec, err := canonicalizeSpec(spec)

79 > if err != nil {

80 return nil, err

81 }

82

83 // load timezone

84 > tz, err := b.loadTimezone(spec) spec.go

85 > if err != nil {

86 return nil, err

87 }

88

89 // compile StructuredCalendarSpecs

90 > ccs := make([]*compiledCalendar, len(spec.StructuredCalendar)) spec.go

91 > for i, structured := range spec.StructuredCalendar {

92 > ccs[i] = newCompiledCalendar(structured, tz) spec.go

93 > }

94

95 // compile excludes

96 > excludes := make([]*compiledCalendar, len(spec.ExcludeStructuredCalendar)) spec.go

97 > for i, excal := range spec.ExcludeStructuredCalendar {

98 > excludes[i] = newCompiledCalendar(excal, tz) spec.go

99 > }

100

101 > cspec := &CompiledSpec{ spec.go

102 > spec: spec,

103 > tz: tz,

104 > calendar: ccs,

105 > excludes: excludes,

106 > warnIterations: b.warnIterations,

107 > maxIterations: b.maxIterations,

108 > }

109 >

110 > return cspec, nil

111 }

112

141

142 //revive:disable-next-line:cognitive-complexity

143 > func canonicalizeSpec(spec *schedulepb.ScheduleSpec) (*schedulepb.ScheduleSpec, error) { spec.go

144 > // make copy so we can change some fields

145 > spec = common.CloneProto(spec)

146 >

147 > // parse CalendarSpecs to StructuredCalendarSpecs

148 > for _, cal := range spec.Calendar {

149 structured, err := parseCalendarToStructured(cal)

150 if err != nil {

153 spec.StructuredCalendar = append(spec.StructuredCalendar, structured)

154 }

155 > spec.Calendar = nil spec.go

156 >

157 > // parse ExcludeCalendars

158 > for _, cal := range spec.ExcludeCalendar {

159 structured, err := parseCalendarToStructured(cal)

160 if err != nil {

163 spec.ExcludeStructuredCalendar = append(spec.ExcludeStructuredCalendar, structured)

164 }

165 > spec.ExcludeCalendar = nil spec.go

166 >

167 > // parse CronStrings

168 > const unset = "__unset__"

169 > cronTZ := unset

170 > for _, cs := range spec.CronString {

171 structured, interval, tz, err := parseCronString(cs)

172 if err != nil {

185 }

186 }

187 > spec.CronString = nil spec.go

188 >

189 > // if we have cron string(s), copy the timezone to spec, checking for conflict first.

190 > // if cron string timezone is empty string, don't copy, let the one in spec be used.

191 > if cronTZ != unset && cronTZ != "" {

192 if spec.TimezoneName != "" && spec.TimezoneName != cronTZ || spec.TimezoneData != nil {

193 return nil, errConflictingTimezoneNames

198

199 // validate structured calendar

200 > for _, structured := range spec.StructuredCalendar { spec.go

201 > if err := validateStructuredCalendar(structured); err != nil { spec.go

202 return nil, err

203 }

205

206 // validate intervals

207 > for _, interval := range spec.Interval { spec.go

208 > if err := validateInterval(interval); err != nil { spec.go

209 return nil, err

210 }

211 }

212

213 > return spec, nil spec.go

214 }

215

216 > func validateStructuredCalendar(scs *schedulepb.StructuredCalendarSpec) error { spec.go

217 > var errs []string

218 >

219 > checkRanges := func(ranges []*schedulepb.Range, field string, minVal, maxVal int32) {

220 > for _, r := range ranges {

221 > if r == nil { // shouldn't happen

222 errs = append(errs, "range is nil")

223 continue

224 }

225 > if r.Start < minVal || r.Start > maxVal { spec.go

226 errs = append(errs, fmt.Sprintf("%s Start is not in range [%d-%d]", field, minVal, maxVal))

227 }

228 > if r.End != 0 && (r.End < r.Start || r.End > maxVal) { spec.go

229 errs = append(errs, fmt.Sprintf("%s End is before Start or not in range [%d-%d]", field, minVal, maxVal))

230 }

231 > if r.Step < 0 { spec.go

232 errs = append(errs, fmt.Sprintf("%s has invalid Step", field))

233 }

235 }

236

237 > checkRanges(scs.Second, "Second", 0, 59) spec.go

238 > checkRanges(scs.Minute, "Minute", 0, 59)

239 > checkRanges(scs.Hour, "Hour", 0, 23)

240 > checkRanges(scs.DayOfMonth, "DayOfMonth", 1, 31)

241 > checkRanges(scs.Month, "Month", 1, 12)

242 > checkRanges(scs.Year, "Year", minCalendarYear, maxCalendarYear)

243 > checkRanges(scs.DayOfWeek, "DayOfWeek", 0, 6)

244 >

245 > if len(scs.Comment) > maxCommentLen {

246 errs = append(errs, "comment is too long")

247 }

248

249 > if len(errs) > 0 { spec.go

250 return errors.New("invalid calendar spec: " + strings.Join(errs, ", "))

251 }

252 > return nil spec.go

253 }

254

255 > func validateInterval(i *schedulepb.IntervalSpec) error { spec.go

256 > if i == nil {

257 return errors.New("interval is nil")

258 }

259 // TODO: use timestamp.ValidateAndCapProtoDuration after switching to state machine based implementation.

260 // Not adding it to workflow based implementation to avoid potential non-determinism errors.

261 > iv, phase := timestamp.DurationValue(i.Interval), timestamp.DurationValue(i.Phase) spec.go

262 > if iv < time.Second {

263 return errors.New("interval is too small")

264 > } else if phase < 0 { spec.go

265 return errors.New("phase is negative")

266 > } else if phase >= iv { spec.go

267 return errors.New("phase cannot be greater than Interval")

268 }

269 > return nil spec.go

270 }

271

272 > func (b *SpecBuilder) loadTimezone(spec *schedulepb.ScheduleSpec) (*time.Location, error) { spec.go

273 > if spec.TimezoneData != nil {

274 return time.LoadLocationFromTZData(spec.TimezoneName, spec.TimezoneData)

275 }

276

277 > if cached, ok := b.locationCache.Get(spec.TimezoneName).(*locationAndError); ok { spec.go

278 > return cached.loc, cached.err spec.go

279 > }

280 > loc, err := time.LoadLocation(spec.TimezoneName) spec.go

281 > b.locationCache.Put(spec.TimezoneName, &locationAndError{

282 > loc: loc,

283 > err: err,

284 > })

285 > return loc, err

286 }

287

293 // Returns: Nominal is the time that matches, pre-jitter. Next is the nominal time with

294 // jitter applied. If there is no matching time, Nominal and Next will be the zero time.

295 > func (cs *CompiledSpec) GetNextTime(jitterSeed string, after time.Time) (GetNextTimeResult, error) { spec.go

296 > // If we're starting before the schedule's allowed time range, jump up to right before

297 > // it (so that we can still return the first second of the range if it happens to match).

298 > // note: AsTime returns unix epoch on nil StartTime

299 > after = util.MaxTime(after, cs.spec.StartTime.AsTime().Add(-time.Second))

300 >

301 > pastEndTime := func(t time.Time) bool {

302 > return cs.spec.EndTime != nil && t.After(cs.spec.EndTime.AsTime()) || t.Year() > maxCalendarYear spec.go

303 > }

304 > warnIterations := cs.warnIterations() spec.go

305 > if warnIterations <= 0 {

306 warnIterations = DefaultWarnIterations

307 }

308 > maxIterations := cs.maxIterations() spec.go

309 > if maxIterations <= 0 {

310 maxIterations = math.MaxInt // disabled: effectively unlimited

311 }

312

313 > var warned bool spec.go

314 > var nominal time.Time

315 > for iterations := 0; nominal.IsZero() || cs.excluded(nominal); iterations++ {

316 > // Hard bound: stop an over-excluded / adversarial spec from spinning toward

317 > // maxCalendarYear. Disabled by default (maxIterations == math.MaxInt); an operator can

318 > // lower it to re-enable enforcement. Well-formed specs resolve in a handful of iterations.

319 > if iterations >= maxIterations {

320 return GetNextTimeResult{ComputeLimitWarning: true}, ErrComputeLimitExceeded

321 }

322 > if iterations >= warnIterations { spec.go

323 warned = true

324 }

325 > nominal = cs.rawNextTime(after) spec.go

326 > after = nominal

327 >

328 > if nominal.IsZero() || pastEndTime(nominal) {

329 return GetNextTimeResult{ComputeLimitWarning: warned}, nil

330 }

331 }

332

333 > maxJitter := timestamp.DurationValue(cs.spec.Jitter) spec.go

334 > // Ensure that jitter doesn't push this time past the _next_ nominal start time

335 > if following := cs.rawNextTime(nominal); !following.IsZero() {

336 > maxJitter = min(maxJitter, following.Sub(nominal)) spec.go

337 > }

338 > next := cs.addJitter(jitterSeed, nominal, maxJitter) spec.go

339 >

340 > return GetNextTimeResult{Nominal: nominal, Next: next, ComputeLimitWarning: warned}, nil

341 }

342

343 // Returns the next matching time (without jitter), or the zero value if no time matches.

344 > func (cs *CompiledSpec) rawNextTime(after time.Time) (nominal time.Time) { spec.go

345 > var minTimestamp int64 = math.MaxInt64 // unix seconds-since-epoch as int64

346 >

347 > for _, cal := range cs.calendar {

348 > if next := cal.next(after); !next.IsZero() { spec.go

349 > nextTs := next.Unix()

350 > if nextTs < minTimestamp {

351 > minTimestamp = nextTs

352 > }

353 }

354 }

355

356 > ts := after.Unix() spec.go

357 > for _, iv := range cs.spec.Interval {

358 > next := cs.nextIntervalTime(iv, ts) spec.go

359 > if next < minTimestamp {

360 > minTimestamp = next

361 > }

362 }

363

364 > if minTimestamp == math.MaxInt64 { spec.go

365 return time.Time{}

366 }

367 > return time.Unix(minTimestamp, 0).UTC() spec.go

368 }

369

370 // Returns the next matching time for a single interval spec.

371 > func (cs *CompiledSpec) nextIntervalTime(iv *schedulepb.IntervalSpec, ts int64) int64 { spec.go

372 > interval := max(int64(timestamp.DurationValue(iv.Interval)/time.Second), 1)

373 > phase := max(int64(timestamp.DurationValue(iv.Phase)/time.Second), 0)

374 > return (((ts-phase)/interval)+1)*interval + phase

375 > }

376

377 // Returns true if any exclude spec matches the time.

378 > func (cs *CompiledSpec) excluded(nominal time.Time) bool { spec.go

379 > for _, excal := range cs.excludes {

380 if excal.matches(nominal) {

381 return true

382 }

383 }

384 > return false spec.go

385 }

386

387 // Adds jitter to a nominal time, deterministically (by hashing the given time and a seed).

388 > func (cs *CompiledSpec) addJitter(seed string, nominal time.Time, maxJitter time.Duration) time.Time { spec.go

389 > if maxJitter < 0 {

390 maxJitter = 0

391 }

392

393 > bin, err := nominal.MarshalBinary() spec.go

394 > if err != nil {

395 return nominal

396 }

397

398 > bin = append(bin, []byte(seed)...) spec.go

399 >

400 > // we want to fit the result of a multiply in 64 bits, and use 32 bits of hash, which

401 > // leaves 32 bits for the range. if we use nanoseconds or microseconds, our range is

402 > // limited to only a few seconds or hours. using milliseconds supports up to 49 days.

403 > fp := uint64(farm.Fingerprint32(bin))

404 > ms := min(uint64(maxJitter.Milliseconds()), math.MaxUint32)

405 > jitter := time.Duration((fp*ms)>>32) * time.Millisecond

406 > return nominal.Add(jitter)

407 }

go.temporal.io/server/api/schedule/v1/message.pb.go 126 covered LOC · 23 ranges

Open complete file

88 func (*BufferedStart) ProtoMessage() {}

89

90 > func (x *BufferedStart) ProtoReflect() protoreflect.Message { message.pb.go

91 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[0]

92 > if x != nil {

93 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

94 if ms.LoadMessageInfo() == nil {

97 return ms

98 }

99 > return mi.MessageOf(x) message.pb.go

100 }

101

126 }

127

128 > func (x *BufferedStart) GetOverlapPolicy() v1.ScheduleOverlapPolicy { message.pb.go

129 > if x != nil {

130 > return x.OverlapPolicy

131 > }

132 return v1.ScheduleOverlapPolicy(0)

133 }

284 func (*InternalState) ProtoMessage() {}

285

286 > func (x *InternalState) ProtoReflect() protoreflect.Message { message.pb.go

287 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[2]

288 > if x != nil {

289 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

290 if ms.LoadMessageInfo() == nil {

293 return ms

294 }

295 > return mi.MessageOf(x) message.pb.go

296 }

297

388 }

389

390 > func (x *StartScheduleArgs) Reset() { message.pb.go

391 > *x = StartScheduleArgs{}

392 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[3]

393 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

394 > ms.StoreMessageInfo(mi)

395 > }

396

397 func (x *StartScheduleArgs) String() string {

401 func (*StartScheduleArgs) ProtoMessage() {}

402

403 > func (x *StartScheduleArgs) ProtoReflect() protoreflect.Message { message.pb.go

404 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[3]

405 > if x != nil {

406 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

407 > if ms.LoadMessageInfo() == nil {

408 > ms.StoreMessageInfo(mi)

409 > }

410 > return ms

411 }

412 return mi.MessageOf(x)

455 }

456

457 > func (x *FullUpdateRequest) Reset() { message.pb.go

458 > *x = FullUpdateRequest{}

459 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[4]

460 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

461 > ms.StoreMessageInfo(mi)

462 > }

463

464 func (x *FullUpdateRequest) String() string {

468 func (*FullUpdateRequest) ProtoMessage() {}

469

470 > func (x *FullUpdateRequest) ProtoReflect() protoreflect.Message { message.pb.go

471 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[4]

472 > if x != nil {

473 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

474 > if ms.LoadMessageInfo() == nil {

475 > ms.StoreMessageInfo(mi)

476 > }

477 > return ms

478 }

479 return mi.MessageOf(x)

590 func (*WatchWorkflowRequest) ProtoMessage() {}

591

592 > func (x *WatchWorkflowRequest) ProtoReflect() protoreflect.Message { message.pb.go

593 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[6]

594 > if x != nil {

595 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

596 > if ms.LoadMessageInfo() == nil {

597 > ms.StoreMessageInfo(mi)

598 > }

599 > return ms

600 }

601 return mi.MessageOf(x)

642 }

643

644 > func (x *WatchWorkflowResponse) Reset() { message.pb.go

645 > *x = WatchWorkflowResponse{}

646 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[7]

647 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

648 > ms.StoreMessageInfo(mi)

649 > }

650

651 func (x *WatchWorkflowResponse) String() string {

655 func (*WatchWorkflowResponse) ProtoMessage() {}

656

657 > func (x *WatchWorkflowResponse) ProtoReflect() protoreflect.Message { message.pb.go

658 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[7]

659 > if x != nil {

660 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

661 > if ms.LoadMessageInfo() == nil {

662 > ms.StoreMessageInfo(mi)

663 > }

664 > return ms

665 }

666 return mi.MessageOf(x)

686 }

687

688 > func (x *WatchWorkflowResponse) GetResult() *v12.Payloads { message.pb.go

689 > if x != nil {

690 > if x, ok := x.ResultFailure.(*WatchWorkflowResponse_Result); ok {

691 return x.Result

692 }

693 }

694 > return nil message.pb.go

695 }

696

697 > func (x *WatchWorkflowResponse) GetFailure() *v13.Failure { message.pb.go

698 > if x != nil {

699 > if x, ok := x.ResultFailure.(*WatchWorkflowResponse_Failure); ok {

700 return x.Failure

701 }

702 }

703 > return nil message.pb.go

704 }

705

787 }

788

789 > func (x *StartWorkflowResponse) Reset() { message.pb.go

790 > *x = StartWorkflowResponse{}

791 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[9]

792 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

793 > ms.StoreMessageInfo(mi)

794 > }

795

796 func (x *StartWorkflowResponse) String() string {

800 func (*StartWorkflowResponse) ProtoMessage() {}

801

802 > func (x *StartWorkflowResponse) ProtoReflect() protoreflect.Message { message.pb.go

803 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[9]

804 > if x != nil {

805 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

806 > if ms.LoadMessageInfo() == nil {

807 > ms.StoreMessageInfo(mi)

808 > }

809 > return ms

810 }

811 return mi.MessageOf(x)

986 }

987

988 > func (x *NextTimeCache) Reset() { message.pb.go

989 > *x = NextTimeCache{}

990 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[12]

991 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

992 > ms.StoreMessageInfo(mi)

993 > }

994

995 func (x *NextTimeCache) String() string {

999 func (*NextTimeCache) ProtoMessage() {}

1000

1001 > func (x *NextTimeCache) ProtoReflect() protoreflect.Message { message.pb.go

1002 > mi := &file_temporal_server_api_schedule_v1_message_proto_msgTypes[12]

1003 > if x != nil {

1004 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

1005 > if ms.LoadMessageInfo() == nil {

1006 > ms.StoreMessageInfo(mi)

1007 > }

1008 > return ms

1009 }

1010 return mi.MessageOf(x)

1023 }

1024

1025 > func (x *NextTimeCache) GetStartTime() *timestamppb.Timestamp { message.pb.go

1026 > if x != nil {

1027 > return x.StartTime

1028 > }

1029 return nil

1030 }

1224 }

1225

1226 > func init() { file_temporal_server_api_schedule_v1_message_proto_init() } message.pb.go

1227 > func file_temporal_server_api_schedule_v1_message_proto_init() {

1228 > if File_temporal_server_api_schedule_v1_message_proto != nil {

1229 return

1230 }

1231 > file_temporal_server_api_schedule_v1_message_proto_msgTypes[7].OneofWrappers = []any{ message.pb.go

1232 > (*WatchWorkflowResponse_Result)(nil),

1233 > (*WatchWorkflowResponse_Failure)(nil),

1234 > }

1235 > type x struct{}

1236 > out := protoimpl.TypeBuilder{

1237 > File: protoimpl.DescBuilder{

1238 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

1239 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_schedule_v1_message_proto_rawDesc), len(file_temporal_server_api_schedule_v1_message_proto_rawDesc)),

1240 > NumEnums: 0,

1241 > NumMessages: 13,

1242 > NumExtensions: 0,

1243 > NumServices: 0,

1244 > },

1245 > GoTypes: file_temporal_server_api_schedule_v1_message_proto_goTypes,

1246 > DependencyIndexes: file_temporal_server_api_schedule_v1_message_proto_depIdxs,

1247 > MessageInfos: file_temporal_server_api_schedule_v1_message_proto_msgTypes,

1248 > }.Build()

1249 > File_temporal_server_api_schedule_v1_message_proto = out.File

1250 > file_temporal_server_api_schedule_v1_message_proto_goTypes = nil

1251 > file_temporal_server_api_schedule_v1_message_proto_depIdxs = nil

1252 }

go.temporal.io/server/common/log/zap_logger.go 113 covered LOC · 29 ranges

Open complete file

58 // NewTestLogger returns a logger for tests

59 // Deprecated: Use testlogger.TestLogger instead.

60 > func NewTestLogger() *zapLogger { zap_logger.go

61 > format := os.Getenv(TestLogFormatEnvVar)

62 > if format == "" {

63 > format = "console"

64 > }

65

66 > logger := BuildZapLogger(Config{ zap_logger.go

67 > Level: os.Getenv(TestLogLevelEnvVar),

68 > Format: format,

69 > Development: true,

70 > })

71 >

72 > // Don't include stack traces for warnings during tests. Only include them for logs with level error and above.

73 > logger = logger.WithOptions(zap.AddStacktrace(zap.ErrorLevel))

74 >

75 > return NewZapLogger(logger)

76 }

77

82

83 // NewZapLogger returns a new zap based logger from zap.Logger

84 > func NewZapLogger(zl *zap.Logger) *zapLogger { zap_logger.go

85 > return &zapLogger{

86 > zl: zl,

87 > skip: skipForZapLogger,

88 > baseZl: zl,

89 > }

90 > }

91

92 // BuildZapLogger builds and returns a new zap.Logger for this logging configuration

93 > func BuildZapLogger(cfg Config) *zap.Logger { zap_logger.go

94 > return buildZapLogger(cfg, true)

95 > }

96

97 > func caller(skip int) string { zap_logger.go

98 > _, path, line, ok := runtime.Caller(skip)

99 > if !ok {

100 return ""

101 }

102 > return path + ":" + strconv.Itoa(line) zap_logger.go

103 }

104

105 > func (l *zapLogger) buildFieldsWithCallAt(tags []tag.Tag) []zap.Field { zap_logger.go

106 > fields := make([]zap.Field, len(tags)+1)

107 > l.fillFields(tags, fields)

108 > fields[len(fields)-1] = zap.String(tag.LoggingCallAtKey, caller(l.skip))

109 > return fields

110 > }

111

112 // fillFields fill fields parameter with fields read from tags. Optimized for performance.

113 > func (l *zapLogger) fillFields(tags []tag.Tag, fields []zap.Field) { zap_logger.go

114 > for i, t := range tags {

115 > if zt, ok := t.(tag.ZapTag); ok { zap_logger.go

116 > fields[i] = zt.Field()

117 > } else {

118 fields[i] = zap.Any(t.Key(), t.Value())

119 }

121 }

122

123 > func setDefaultMsg(msg string) string { zap_logger.go

124 > if msg == "" {

125 return defaultMsgForEmpty

126 }

127 > return msg zap_logger.go

128 }

129

130 > func (l *zapLogger) Debug(msg string, tags ...tag.Tag) { zap_logger.go

131 > if l.zl.Core().Enabled(zap.DebugLevel) {

132 msg = setDefaultMsg(msg)

133 fields := l.buildFieldsWithCallAt(tags)

136 }

137

138 > func (l *zapLogger) Info(msg string, tags ...tag.Tag) { zap_logger.go

139 > if l.zl.Core().Enabled(zap.InfoLevel) {

140 > msg = setDefaultMsg(msg)

141 > fields := l.buildFieldsWithCallAt(tags)

142 > l.zl.Info(msg, fields...)

143 > }

144 }

145

191 //

192 // by deduping "foo" against any existing "foo" tags *only in the former*

193 > func (l *zapLogger) With(tags ...tag.Tag) Logger { zap_logger.go

194 > cloneTags := mergeTags(l.tags, tags)

195 > if l.baseZl == nil {

196 l.baseZl = l.zl

197 }

198 > return l.cloneWithTags(cloneTags) zap_logger.go

199 }

200

201 > func (l *zapLogger) cloneWithTags(tags []tag.Tag) Logger { zap_logger.go

202 > fields := make([]zap.Field, len(tags))

203 > l.fillFields(tags, fields)

204 > zl := l.baseZl.With(fields...)

205 > return &zapLogger{

206 > zl: zl,

207 > skip: l.skip,

208 > baseZl: l.baseZl,

209 > tags: tags,

210 > }

211 > }

212

213 > func (l *zapLogger) Skip(extraSkip int) Logger { zap_logger.go

214 > return &zapLogger{

215 > zl: l.zl,

216 > skip: l.skip + extraSkip,

217 > baseZl: l.baseZl,

218 > }

219 > }

220

221 > func mergeTags(oldTags, newTags []tag.Tag) (outTags []tag.Tag) { zap_logger.go

222 > // Even if oldTags empty, we don't just return newTags because we need to de-dupe it.

223 > outTags = slices.Clone(oldTags)

224 > for _, t := range newTags {

225 > if i := slices.IndexFunc(outTags, func(ti tag.Tag) bool {

226 > return ti.Key() == t.Key() zap_logger.go

227 > }); i >= 0 {

228 outTags[i] = t

229 > } else { zap_logger.go

230 > outTags = append(outTags, t)

231 > }

232 }

233 > return outTags zap_logger.go

234 }

235

236 > func buildZapLogger(cfg Config, disableCaller bool) *zap.Logger { zap_logger.go

237 > encodeConfig := DefaultZapEncoderConfig

238 > if disableCaller {

239 > encodeConfig.CallerKey = zapcore.OmitKey

240 > encodeConfig.EncodeCaller = nil

241 > }

242

243 > outputPath := "stderr" zap_logger.go

244 > if len(cfg.OutputFile) > 0 {

245 outputPath = cfg.OutputFile

246 }

247 > if cfg.Stdout { zap_logger.go

248 outputPath = "stdout"

249 }

250 > encoding := "json" zap_logger.go

251 > if cfg.Format == "console" {

252 > encoding = "console" zap_logger.go

253 > }

254 > config := zap.Config{ zap_logger.go

255 > Level: zap.NewAtomicLevelAt(ParseZapLevel(cfg.Level)),

256 > Development: cfg.Development,

257 > Sampling: nil,

258 > Encoding: encoding,

259 > EncoderConfig: encodeConfig,

260 > OutputPaths: []string{outputPath},

261 > ErrorOutputPaths: []string{outputPath},

262 > DisableCaller: disableCaller,

263 > }

264 > logger, _ := config.Build()

265 > return logger

266 }

267

297 }

298

299 > func ParseZapLevel(level string) zapcore.Level { zap_logger.go

300 > switch strings.ToLower(level) {

301 case "debug":

302 return zap.DebugLevel

313 case "fatal":

314 return zap.FatalLevel

315 > default: zap_logger.go

316 > return zap.InfoLevel

317 }

318 }

go.temporal.io/server/common/cache/lru.go 100 covered LOC · 28 ranges

Open complete file

135

136 // New creates a new cache with the given options

137 > func New(maxSize int, opts *Options) StoppableCache { lru.go

138 > return NewWithMetrics(maxSize, opts, metrics.NoopMetricsHandler)

139 > }

140

141 // NewWithMetrics creates a new cache that will emit capacity and ttl metrics.

142 // handler should be tagged with metrics.CacheTypeTag.

143 > func NewWithMetrics(maxSize int, opts *Options, handler metrics.Handler) StoppableCache { lru.go

144 > if opts == nil {

145 opts = &Options{}

146 }

147

148 > backgroundEvict := opts.BackgroundEvict lru.go

149 > if backgroundEvict == nil {

150 > backgroundEvict = func() dynamicconfig.CacheBackgroundEvictSettings { lru.go

151 > return dynamicconfig.CacheBackgroundEvictSettings{

152 > Enabled: false,

153 > }

154 > }

155 }

156

157 > timeSource := opts.TimeSource lru.go

158 > if timeSource == nil {

159 > timeSource = clock.NewRealTimeSource() lru.go

160 > }

161

162 > metrics.CacheSize.With(handler).Record(float64(maxSize)) lru.go

163 > metrics.CacheTtl.With(handler).Record(opts.TTL)

164 > c := &lru{

165 > byAccess: list.New(),

166 > byKey: make(map[any]*list.Element),

167 > ttl: opts.TTL,

168 > maxSize: maxSize,

169 > currSize: 0,

170 > pin: opts.Pin,

171 > onPut: opts.OnPut,

172 > onEvict: opts.OnEvict,

173 > timeSource: timeSource,

174 > metricsHandler: handler,

175 > backgroundEvict: backgroundEvict,

176 > }

177 > if c.backgroundEvict().Enabled {

178 c.loops.Go(c.bgEvictLoop)

179 }

180 > return c lru.go

181 }

182

188

189 // Get retrieves the value stored under the given key

190 > func (c *lru) Get(key any) any { lru.go

191 > if c.maxSize == 0 { //

192 return nil

193 }

194 > c.mut.Lock() lru.go

195 > defer c.mut.Unlock()

196 >

197 > element := c.byKey[key]

198 > if element == nil {

199 > return nil lru.go

200 > }

201

202 > entry := element.Value.(*entryImpl) lru.go

203 >

204 > if c.isEntryExpired(entry, c.timeSource.Now().UTC()) {

205 // Entry has expired

206 c.deleteInternal(element)

208 }

209

210 > metrics.CacheEntryAgeOnGet.With(c.metricsHandler).Record(c.timeSource.Now().UTC().Sub(entry.createTime)) lru.go

211 >

212 > c.updateEntryRefCount(entry)

213 > c.byAccess.MoveToFront(element)

214 > return entry.value

215 }

216

217 // Put puts a new value associated with a given key, returning the existing value (if present)

218 > func (c *lru) Put(key any, value any) any { lru.go

219 > if c.pin {

220 panic("Cannot use Put API in Pin mode. Use Delete and PutIfNotExist if necessary")

221 }

222 > val, _ := c.putInternal(key, value, true) lru.go

223 > return val

224 }

225

294 // Put puts a new value associated with a given key, returning the existing value (if present)

295 // allowUpdate flag is used to control overwrite behavior if the value exists.

296 > func (c *lru) putInternal(key any, value any, allowUpdate bool) (any, error) { lru.go

297 > if c.maxSize == 0 {

298 return nil, nil

299 }

300 > newEntrySize := getSize(value) lru.go

301 > if newEntrySize > c.maxSize {

302 return nil, ErrCacheItemTooLarge

303 }

304

305 > c.mut.Lock() lru.go

306 > defer c.mut.Unlock()

307 >

308 > elt := c.byKey[key]

309 > // If the entry exists, check if it has expired or update the value

310 > if elt != nil {

311 existingEntry := elt.Value.(*entryImpl)

312 if !c.isEntryExpired(existingEntry, c.timeSource.Now().UTC()) {

347 }

348

349 > c.tryEvictUntilEnoughSpaceWithSkipEntry(newEntrySize, nil) lru.go

350 >

351 > // check if the new entry can fit in the cache

352 > newCacheSize := c.calculateNewCacheSize(newEntrySize, emptyEntrySize)

353 > if newCacheSize > c.maxSize {

354 return nil, ErrCacheFull

355 }

356

357 > entry := &entryImpl{ lru.go

358 > key: key,

359 > value: value,

360 > size: newEntrySize,

361 > }

362 > c.updateEntryTTL(entry)

363 > c.updateEntryRefCount(entry)

364 > element := c.byAccess.PushFront(entry)

365 > c.byKey[key] = element

366 > c.currSize = newCacheSize

367 > metrics.CacheUsage.With(c.metricsHandler).Record(float64(c.currSize))

368 >

369 > if c.onPut != nil {

370 c.onPut(value)

371 }

372

373 > return nil, nil lru.go

374 }

375

376 > func (c *lru) calculateNewCacheSize(newEntrySize int, existingEntrySize int) int { lru.go

377 > return c.currSize - existingEntrySize + newEntrySize

378 > }

379

380 func (c *lru) deleteInternal(element *list.Element) {

397 // tryEvictUntilEnoughSpaceWithSkipEntry try to evict entries until there is enough space for the new entry without

398 // evicting the existing entry. the existing entry is skipped because it is being updated.

399 > func (c *lru) tryEvictUntilEnoughSpaceWithSkipEntry(newEntrySize int, existingEntry *entryImpl) { lru.go

400 > element := c.byAccess.Back()

401 > existingEntrySize := 0

402 > if existingEntry != nil {

403 existingEntrySize = existingEntry.Size()

404 }

405

406 > for c.calculateNewCacheSize(newEntrySize, existingEntrySize) > c.maxSize && element != nil { lru.go

407 entry := element.Value.(*entryImpl)

408 if existingEntry != nil && entry.key == existingEntry.key {

426 }

427

428 > func (c *lru) isEntryExpired(entry *entryImpl, currentTime time.Time) bool { lru.go

429 > return entry.refCount == 0 && !entry.createTime.IsZero() && currentTime.After(entry.createTime.Add(c.ttl))

430 > }

431

432 > func (c *lru) updateEntryTTL(entry *entryImpl) { lru.go

433 > if c.ttl != 0 {

434 > entry.createTime = c.timeSource.Now().UTC() lru.go

435 > }

436 }

437

438 > func (c *lru) updateEntryRefCount(entry *entryImpl) { lru.go

439 > if c.pin {

440 entry.refCount++

441 if entry.refCount == 1 {

go.temporal.io/server/api/persistence/v1/predicates.pb.go 82 covered LOC · 22 ranges

Open complete file

57 func (*Predicate) ProtoMessage() {}

58

59 > func (x *Predicate) ProtoReflect() protoreflect.Message { predicates.pb.go

60 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0]

61 > if x != nil {

62 > ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

63 > if ms.LoadMessageInfo() == nil {

64 > ms.StoreMessageInfo(mi)

65 > }

66 > return ms

67 }

68 return mi.MessageOf(x)

261 func (*UniversalPredicateAttributes) ProtoMessage() {}

262

263 > func (x *UniversalPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

264 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[1]

265 > if x != nil {

266 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

267 if ms.LoadMessageInfo() == nil {

270 return ms

271 }

272 > return mi.MessageOf(x) predicates.pb.go

273 }

274

297 func (*EmptyPredicateAttributes) ProtoMessage() {}

298

299 > func (x *EmptyPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

300 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[2]

301 > if x != nil {

302 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

303 if ms.LoadMessageInfo() == nil {

306 return ms

307 }

308 > return mi.MessageOf(x) predicates.pb.go

309 }

310

334 func (*AndPredicateAttributes) ProtoMessage() {}

335

336 > func (x *AndPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

337 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[3]

338 > if x != nil {

339 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

340 if ms.LoadMessageInfo() == nil {

343 return ms

344 }

345 > return mi.MessageOf(x) predicates.pb.go

346 }

347

378 func (*OrPredicateAttributes) ProtoMessage() {}

379

380 > func (x *OrPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

381 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[4]

382 > if x != nil {

383 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

384 if ms.LoadMessageInfo() == nil {

387 return ms

388 }

389 > return mi.MessageOf(x) predicates.pb.go

390 }

391

422 func (*NotPredicateAttributes) ProtoMessage() {}

423

424 > func (x *NotPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

425 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[5]

426 > if x != nil {

427 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

428 if ms.LoadMessageInfo() == nil {

431 return ms

432 }

433 > return mi.MessageOf(x) predicates.pb.go

434 }

435

466 func (*NamespaceIdPredicateAttributes) ProtoMessage() {}

467

468 > func (x *NamespaceIdPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

469 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[6]

470 > if x != nil {

471 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

472 if ms.LoadMessageInfo() == nil {

475 return ms

476 }

477 > return mi.MessageOf(x) predicates.pb.go

478 }

479

510 func (*TaskTypePredicateAttributes) ProtoMessage() {}

511

512 > func (x *TaskTypePredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

513 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[7]

514 > if x != nil {

515 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

516 if ms.LoadMessageInfo() == nil {

519 return ms

520 }

521 > return mi.MessageOf(x) predicates.pb.go

522 }

523

554 func (*DestinationPredicateAttributes) ProtoMessage() {}

555

556 > func (x *DestinationPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

557 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[8]

558 > if x != nil {

559 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

560 if ms.LoadMessageInfo() == nil {

563 return ms

564 }

565 > return mi.MessageOf(x) predicates.pb.go

566 }

567

598 func (*OutboundTaskGroupPredicateAttributes) ProtoMessage() {}

599

600 > func (x *OutboundTaskGroupPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

601 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[9]

602 > if x != nil {

603 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

604 if ms.LoadMessageInfo() == nil {

607 return ms

608 }

609 > return mi.MessageOf(x) predicates.pb.go

610 }

611

642 func (*OutboundTaskPredicateAttributes) ProtoMessage() {}

643

644 > func (x *OutboundTaskPredicateAttributes) ProtoReflect() protoreflect.Message { predicates.pb.go

645 > mi := &file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[10]

646 > if x != nil {

647 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))

648 if ms.LoadMessageInfo() == nil {

651 return ms

652 }

653 > return mi.MessageOf(x) predicates.pb.go

654 }

655

828 }

829

830 > func init() { file_temporal_server_api_persistence_v1_predicates_proto_init() } predicates.pb.go

831 > func file_temporal_server_api_persistence_v1_predicates_proto_init() {

832 > if File_temporal_server_api_persistence_v1_predicates_proto != nil {

833 > return

834 > }

835 > file_temporal_server_api_persistence_v1_predicates_proto_msgTypes[0].OneofWrappers = []any{

836 > (*Predicate_UniversalPredicateAttributes)(nil),

837 > (*Predicate_EmptyPredicateAttributes)(nil),

838 > (*Predicate_AndPredicateAttributes)(nil),

839 > (*Predicate_OrPredicateAttributes)(nil),

840 > (*Predicate_NotPredicateAttributes)(nil),

841 > (*Predicate_NamespaceIdPredicateAttributes)(nil),

842 > (*Predicate_TaskTypePredicateAttributes)(nil),

843 > (*Predicate_DestinationPredicateAttributes)(nil),

844 > (*Predicate_OutboundTaskGroupPredicateAttributes)(nil),

845 > (*Predicate_OutboundTaskPredicateAttributes)(nil),

846 > }

847 > type x struct{}

848 > out := protoimpl.TypeBuilder{

849 > File: protoimpl.DescBuilder{

850 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

851 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc), len(file_temporal_server_api_persistence_v1_predicates_proto_rawDesc)),

852 > NumEnums: 0,

853 > NumMessages: 12,

854 > NumExtensions: 0,

855 > NumServices: 0,

856 > },

857 > GoTypes: file_temporal_server_api_persistence_v1_predicates_proto_goTypes,

858 > DependencyIndexes: file_temporal_server_api_persistence_v1_predicates_proto_depIdxs,

859 > MessageInfos: file_temporal_server_api_persistence_v1_predicates_proto_msgTypes,

860 > }.Build()

861 > File_temporal_server_api_persistence_v1_predicates_proto = out.File

862 > file_temporal_server_api_persistence_v1_predicates_proto_goTypes = nil

863 > file_temporal_server_api_persistence_v1_predicates_proto_depIdxs = nil

864 }

go.temporal.io/server/chasm/search_attribute.go 75 covered LOC · 12 ranges

Open complete file

136 }

137

138 > func newSearchAttributeFieldBool(index int) SearchAttributeFieldBool { search_attribute.go

139 > return SearchAttributeFieldBool{

140 > field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_BOOL, index),

141 > }

142 > }

143

144 // SearchAttributeFieldDateTime is a search attribute field for a datetime value.

147 }

148

149 > func newSearchAttributeFieldDateTime(index int) SearchAttributeFieldDateTime { search_attribute.go

150 > return SearchAttributeFieldDateTime{

151 > field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_DATETIME, index),

152 > }

153 > }

154

155 // SearchAttributeFieldInt is a search attribute field for an integer value.

158 }

159

160 > func newSearchAttributeFieldInt(index int) SearchAttributeFieldInt { search_attribute.go

161 > return SearchAttributeFieldInt{

162 > field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_INT, index),

163 > }

164 > }

165

166 // SearchAttributeFieldDouble is a search attribute field for a double value.

169 }

170

171 > func newSearchAttributeFieldDouble(index int) SearchAttributeFieldDouble { search_attribute.go

172 > return SearchAttributeFieldDouble{

173 > field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_DOUBLE, index),

174 > }

175 > }

176

177 // SearchAttributeFieldKeyword is a search attribute field for a keyword value.

180 }

181

182 > func newSearchAttributeFieldKeyword(index int) SearchAttributeFieldKeyword { search_attribute.go

183 > return SearchAttributeFieldKeyword{

184 > field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_KEYWORD, index),

185 > }

186 > }

187

188 > func newSearchAttributeFieldLowCardinalityKeyword(index int) SearchAttributeFieldKeyword { search_attribute.go

189 > return SearchAttributeFieldKeyword{

190 > field: fmt.Sprintf("%s%s%02d", sadefs.ReservedPrefix, "LowCardinalityKeyword", index),

191 > }

192 > }

193

194 // SearchAttributeFieldKeywordList is a search attribute field for a keyword list value.

197 }

198

199 > func newSearchAttributeFieldKeywordList(index int) SearchAttributeFieldKeywordList { search_attribute.go

200 > return SearchAttributeFieldKeywordList{

201 > field: resolveFieldName(enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST, index),

202 > }

203 > }

204

205 // SearchAttributeFieldText is a search attribute field for a text value.

214 }

215

216 > func resolveFieldName(valueType enumspb.IndexedValueType, index int) string { search_attribute.go

217 > // Columns are named like TemporalBool01, TemporalDatetime01, TemporalDouble01, TemporalInt01.

218 > return fmt.Sprintf("%s%s%02d", sadefs.ReservedPrefix, valueType.String(), index)

219 > }

220

221 func (s searchAttributeDefinition) definition() searchAttributeDefinition {

239 }

240

241 > func newSearchAttributeBoolByField(field string) SearchAttributeBool { search_attribute.go

242 > return SearchAttributeBool{

243 > searchAttributeDefinition: searchAttributeDefinition{

244 > alias: field,

245 > field: field,

246 > valueType: enumspb.INDEXED_VALUE_TYPE_BOOL,

247 > },

248 > }

249 > }

250

251 // Value sets the boolean value of the search attribute.

276 }

277

278 > func newSearchAttributeDateTimeByField(field string) SearchAttributeDateTime { search_attribute.go

279 > return SearchAttributeDateTime{

280 > searchAttributeDefinition: searchAttributeDefinition{

281 > alias: field,

282 > field: field,

283 > valueType: enumspb.INDEXED_VALUE_TYPE_DATETIME,

284 > },

285 > }

286 > }

287

288 // Value sets the date time value of the search attribute.

377 }

378

379 > func newSearchAttributeKeywordByField(field string) SearchAttributeKeyword { search_attribute.go

380 > return SearchAttributeKeyword{

381 > searchAttributeDefinition: searchAttributeDefinition{

382 > alias: field,

383 > field: field,

384 > valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD,

385 > },

386 > }

387 > }

388

389 // Value sets the string value of the search attribute.

414 }

415

416 > func newSearchAttributeKeywordListByField(field string) SearchAttributeKeywordList { search_attribute.go

417 > return SearchAttributeKeywordList{

418 > searchAttributeDefinition: searchAttributeDefinition{

419 > alias: field,

420 > field: field,

421 > valueType: enumspb.INDEXED_VALUE_TYPE_KEYWORD_LIST,

422 > },

423 > }

424 > }

425

426 // Value sets the string list value of the search attribute.

go.temporal.io/server/service/worker/scheduler/calendar.go 71 covered LOC · 25 ranges

Open complete file

80 )

81

82 > func newCompiledCalendar(cal *schedulepb.StructuredCalendarSpec, tz *time.Location) *compiledCalendar { calendar.go

83 > return &compiledCalendar{

84 > tz: tz,

85 > year: makeYearMatcher(cal.Year),

86 > month: makeBitMatcher(cal.Month),

87 > dayOfMonth: makeBitMatcher(cal.DayOfMonth),

88 > dayOfWeek: makeBitMatcher(cal.DayOfWeek),

89 > hour: makeBitMatcher(cal.Hour),

90 > minute: makeBitMatcher(cal.Minute),

91 > second: makeBitMatcher(cal.Second),

92 > }

93 > }

94

95 // Returns true if the given time matches this calendar spec.

113 // Returns the earliest time that matches this calendar spec that is after the given time.

114 // All times are considered to have 1 second resolution.

115 > func (cc *compiledCalendar) next(ts time.Time) time.Time { calendar.go

116 > // set time zone

117 > ts = ts.In(cc.tz)

118 >

119 > // get ymdhms from ts

120 > y, mo, d := ts.Date()

121 > h, m, s := ts.Clock()

122 > dstoffset := 0 * time.Hour

123 > if ts.Add(-time.Hour).Hour() == h {

124 // we're in the second copy of a dst repeated hour

125 dstoffset = 1 * time.Hour

127

128 // looking for first matching time after ts, so add 1 second

129 > s++ calendar.go

130 >

131 > Outer:

132 > for {

133 > // normalize after carries

134 > if s >= 60 {

135 > m, s = m+1, 0 calendar.go

136 > }

137 > if m >= 60 { calendar.go

138 prev := time.Date(y, mo, d, h, 0, 0, 0, cc.tz)

139 h, m = h+1, 0

148 }

149 }

150 > if h >= 24 { calendar.go

151 d, h = d+1, 0

152 }

153 > if d > daysInMonth(mo, y) { calendar.go

154 mo, d = mo+1, 1

155 }

156 > if mo > time.December { calendar.go

157 y, mo = y+1, time.January

158 }

159 > if y > maxCalendarYear { calendar.go

160 break Outer

161 }

162 // try to match year, month, etc. from outside in

163 > if !cc.year(y) { calendar.go

164 y, mo, d, h, m, s = y+1, time.January, 1, 0, 0, 0

165 dstoffset = 0

166 continue Outer

167 }

168 > for !cc.month(int(mo)) { calendar.go

169 mo, d, h, m, s = mo+1, 1, 0, 0, 0

170 dstoffset = 0

173 }

174 }

175 > for !cc.dayOfMonth(d) || !cc.dayOfWeek(int(time.Date(y, mo, d, h, m, s, 0, cc.tz).Weekday())) { calendar.go

176 d, h, m, s = d+1, 0, 0, 0

177 dstoffset = 0

180 }

181 }

182 > for !cc.hour(h) { calendar.go

183 h, m, s = h+1, 0, 0

184 dstoffset = 0

187 }

188 }

189 > for !cc.minute(m) { calendar.go

190 m, s = m+1, 0

191 if m >= 60 {

193 }

194 }

195 > for !cc.second(s) { calendar.go

196 > s = s + 1 calendar.go

197 > if s >= 60 {

198 > continue Outer

199 }

200 }

201 // everything matches

202 > nextTs := time.Date(y, mo, d, h, m, s, 0, cc.tz) calendar.go

203 > // we might have reached a nonexistent time that got jumped over by a dst transition.

204 > // we can tell if the hour is different from what we think it should be.

205 > if nextTs.Hour() != h {

206 h, m, s = h+1, 0, 0

207 continue Outer

208 }

209 > return nextTs.Add(dstoffset) calendar.go

210 }

211

342 }

343

344 > func makeBitMatcher(ranges []*schedulepb.Range) func(int) bool { calendar.go

345 > var bits uint64

346 > add := func(i int) { bits |= 1 << i }

347 > iterateRanges(ranges, add)

348 > return func(v int) bool { return (1<<v)&bits != 0 }

349 }

350

351 > func makeYearMatcher(ranges []*schedulepb.Range) func(int) bool { calendar.go

352 > if len(ranges) == 0 {

353 > // special case for year: all is represented as empty range list calendar.go

354 > return func(int) bool { return true }

355 }

356

368 }

369

370 > func iterateRanges(ranges []*schedulepb.Range, f func(i int)) { calendar.go

371 > for _, r := range ranges {

372 > start, end, step := int(r.GetStart()), int(r.GetEnd()), int(r.GetStep())

373 > if step == 0 {

374 > step = 1

375 > }

376 > if end < start {

377 end = start

378 }

379 > for ; start <= end; start += step { calendar.go

380 > f(start)

381 > }

382 }

383 }

548 }

549

550 > func daysInMonth(m time.Month, y int) int { calendar.go

551 > if m == time.February {

552 if isLeapYear(y) {

553 return 29

556 }

557 }

558 > const bits = 0b1010110101010 calendar.go

559 > return 30 + (bits>>m)&1

560 }

go.temporal.io/server/api/persistence/v1/executions.pb.go 60 covered LOC · 1 range

Open complete file

5706 }

5707

5708 > func init() { file_temporal_server_api_persistence_v1_executions_proto_init() } executions.pb.go

5709 > func file_temporal_server_api_persistence_v1_executions_proto_init() {

5710 > if File_temporal_server_api_persistence_v1_executions_proto != nil {

5711 > return

5712 > }

5713 > file_temporal_server_api_persistence_v1_chasm_proto_init()

5714 > file_temporal_server_api_persistence_v1_hsm_proto_init()

5715 > file_temporal_server_api_persistence_v1_queues_proto_init()

5716 > file_temporal_server_api_persistence_v1_update_proto_init()

5717 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[1].OneofWrappers = []any{

5718 > (*WorkflowExecutionInfo_LastWorkflowTaskFailureCause)(nil),

5719 > (*WorkflowExecutionInfo_LastWorkflowTaskTimedOutType)(nil),

5720 > }

5721 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[8].OneofWrappers = []any{

5722 > (*TransferTaskInfo_CloseExecutionTaskDetails_)(nil),

5723 > (*TransferTaskInfo_ChasmTaskInfo)(nil),

5724 > }

5725 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[10].OneofWrappers = []any{

5726 > (*VisibilityTaskInfo_ChasmTaskInfo)(nil),

5727 > }

5728 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[11].OneofWrappers = []any{

5729 > (*TimerTaskInfo_ChasmTaskInfo)(nil),

5730 > }

5731 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[13].OneofWrappers = []any{

5732 > (*OutboundTaskInfo_StateMachineInfo)(nil),

5733 > (*OutboundTaskInfo_ChasmTaskInfo)(nil),

5734 > (*OutboundTaskInfo_WorkerCommandsTask)(nil),

5735 > }

5736 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[17].OneofWrappers = []any{

5737 > (*ActivityInfo_UseWorkflowBuildIdInfo_)(nil),

5738 > (*ActivityInfo_LastIndependentlyAssignedBuildId)(nil),

5739 > }

5740 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[23].OneofWrappers = []any{

5741 > (*Callback_Nexus_)(nil),

5742 > (*Callback_Hsm)(nil),

5743 > }

5744 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[40].OneofWrappers = []any{

5745 > (*ActivityInfo_PauseInfo_Manual_)(nil),

5746 > (*ActivityInfo_PauseInfo_RuleId)(nil),

5747 > }

5748 > file_temporal_server_api_persistence_v1_executions_proto_msgTypes[46].OneofWrappers = []any{

5749 > (*CallbackInfo_Trigger_WorkflowClosed)(nil),

5750 > }

5751 > type x struct{}

5752 > out := protoimpl.TypeBuilder{

5753 > File: protoimpl.DescBuilder{

5754 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

5755 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_executions_proto_rawDesc), len(file_temporal_server_api_persistence_v1_executions_proto_rawDesc)),

5756 > NumEnums: 0,

5757 > NumMessages: 47,

5758 > NumExtensions: 0,

5759 > NumServices: 0,

5760 > },

5761 > GoTypes: file_temporal_server_api_persistence_v1_executions_proto_goTypes,

5762 > DependencyIndexes: file_temporal_server_api_persistence_v1_executions_proto_depIdxs,

5763 > MessageInfos: file_temporal_server_api_persistence_v1_executions_proto_msgTypes,

5764 > }.Build()

5765 > File_temporal_server_api_persistence_v1_executions_proto = out.File

5766 > file_temporal_server_api_persistence_v1_executions_proto_goTypes = nil

5767 > file_temporal_server_api_persistence_v1_executions_proto_depIdxs = nil

5768 }

go.temporal.io/server/api/historyservice/v1/request_response.pb.go 45 covered LOC · 1 range

Open complete file

11956 }

11957

11958 > func init() { file_temporal_server_api_historyservice_v1_request_response_proto_init() } request_response.pb.go

11959 > func file_temporal_server_api_historyservice_v1_request_response_proto_init() {

11960 > if File_temporal_server_api_historyservice_v1_request_response_proto != nil {

11961 > return

11962 > }

11963 > file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[107].OneofWrappers = []any{

11964 > (*StreamWorkflowReplicationMessagesRequest_SyncReplicationState)(nil),

11965 > }

11966 > file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[108].OneofWrappers = []any{

11967 > (*StreamWorkflowReplicationMessagesResponse_Messages)(nil),

11968 > }

11969 > file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[134].OneofWrappers = []any{

11970 > (*CompleteNexusOperationChasmRequest_Success)(nil),

11971 > (*CompleteNexusOperationChasmRequest_Failure)(nil),

11972 > }

11973 > file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[136].OneofWrappers = []any{

11974 > (*CompleteNexusOperationRequest_Success)(nil),

11975 > (*CompleteNexusOperationRequest_Failure)(nil),

11976 > }

11977 > file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[162].OneofWrappers = []any{

11978 > (*ExecuteMultiOperationRequest_Operation_StartWorkflow)(nil),

11979 > (*ExecuteMultiOperationRequest_Operation_UpdateWorkflow)(nil),

11980 > }

11981 > file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes[163].OneofWrappers = []any{

11982 > (*ExecuteMultiOperationResponse_Response_StartWorkflow)(nil),

11983 > (*ExecuteMultiOperationResponse_Response_UpdateWorkflow)(nil),

11984 > }

11985 > type x struct{}

11986 > out := protoimpl.TypeBuilder{

11987 > File: protoimpl.DescBuilder{

11988 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

11989 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_historyservice_v1_request_response_proto_rawDesc)),

11990 > NumEnums: 0,

11991 > NumMessages: 171,

11992 > NumExtensions: 1,

11993 > NumServices: 0,

11994 > },

11995 > GoTypes: file_temporal_server_api_historyservice_v1_request_response_proto_goTypes,

11996 > DependencyIndexes: file_temporal_server_api_historyservice_v1_request_response_proto_depIdxs,

11997 > MessageInfos: file_temporal_server_api_historyservice_v1_request_response_proto_msgTypes,

11998 > ExtensionInfos: file_temporal_server_api_historyservice_v1_request_response_proto_extTypes,

11999 > }.Build()

12000 > File_temporal_server_api_historyservice_v1_request_response_proto = out.File

12001 > file_temporal_server_api_historyservice_v1_request_response_proto_goTypes = nil

12002 > file_temporal_server_api_historyservice_v1_request_response_proto_depIdxs = nil

12003 }

go.temporal.io/server/api/matchingservice/v1/request_response.pb.go 43 covered LOC · 1 range

Open complete file

6833 }

6834

6835 > func init() { file_temporal_server_api_matchingservice_v1_request_response_proto_init() } request_response.pb.go

6836 > func file_temporal_server_api_matchingservice_v1_request_response_proto_init() {

6837 > if File_temporal_server_api_matchingservice_v1_request_response_proto != nil {

6838 > return

6839 > }

6840 > file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[27].OneofWrappers = []any{

6841 > (*UpdateWorkerBuildIdCompatibilityRequest_ApplyPublicRequest_)(nil),

6842 > (*UpdateWorkerBuildIdCompatibilityRequest_RemoveBuildIds_)(nil),

6843 > (*UpdateWorkerBuildIdCompatibilityRequest_PersistUnknownBuildId)(nil),

6844 > }

6845 > file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[29].OneofWrappers = []any{

6846 > (*GetWorkerVersioningRulesRequest_Request)(nil),

6847 > }

6848 > file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[31].OneofWrappers = []any{

6849 > (*UpdateWorkerVersioningRulesRequest_Request)(nil),

6850 > }

6851 > file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[37].OneofWrappers = []any{

6852 > (*SyncDeploymentUserDataRequest_UpdateVersionData)(nil),

6853 > (*SyncDeploymentUserDataRequest_ForgetVersion)(nil),

6854 > }

6855 > file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes[56].OneofWrappers = []any{

6856 > (*DispatchNexusTaskResponse_HandlerError)(nil),

6857 > (*DispatchNexusTaskResponse_Response)(nil),

6858 > (*DispatchNexusTaskResponse_RequestTimeout)(nil),

6859 > (*DispatchNexusTaskResponse_Failure)(nil),

6860 > }

6861 > type x struct{}

6862 > out := protoimpl.TypeBuilder{

6863 > File: protoimpl.DescBuilder{

6864 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

6865 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_matchingservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_matchingservice_v1_request_response_proto_rawDesc)),

6866 > NumEnums: 0,

6867 > NumMessages: 97,

6868 > NumExtensions: 0,

6869 > NumServices: 0,

6870 > },

6871 > GoTypes: file_temporal_server_api_matchingservice_v1_request_response_proto_goTypes,

6872 > DependencyIndexes: file_temporal_server_api_matchingservice_v1_request_response_proto_depIdxs,

6873 > MessageInfos: file_temporal_server_api_matchingservice_v1_request_response_proto_msgTypes,

6874 > }.Build()

6875 > File_temporal_server_api_matchingservice_v1_request_response_proto = out.File

6876 > file_temporal_server_api_matchingservice_v1_request_response_proto_goTypes = nil

6877 > file_temporal_server_api_matchingservice_v1_request_response_proto_depIdxs = nil

6878 }

go.temporal.io/server/common/metrics/defs.go 41 covered LOC · 7 ranges

Open complete file

20 )

21

22 > func NewTimerDef(name string, opts ...Option) timerDefinition { defs.go

23 > // This line cannot be combined with others!

24 > // This ensures the stack trace has information of the caller.

25 > def := newMetricDefinition(name, opts...)

26 > globalRegistry.register(def)

27 > return timerDefinition{def}

28 > }

29

30 > func NewBytesHistogramDef(name string, opts ...Option) histogramDefinition { defs.go

31 > // This line cannot be combined with others!

32 > // This ensures the stack trace has information of the caller.

33 > def := newMetricDefinition(name, append(opts, WithUnit(Bytes))...)

34 > globalRegistry.register(def)

35 > return histogramDefinition{def}

36 > }

37

38 > func NewDimensionlessHistogramDef(name string, opts ...Option) histogramDefinition { defs.go

39 > // This line cannot be combined with others!

40 > // This ensures the stack trace has information of the caller.

41 > def := newMetricDefinition(name, append(opts, WithUnit(Dimensionless))...)

42 > globalRegistry.register(def)

43 > return histogramDefinition{def}

44 > }

45

46 > func NewCounterDef(name string, opts ...Option) counterDefinition { defs.go

47 > // This line cannot be combined with others!

48 > // This ensures the stack trace has information of the caller.

49 > def := newMetricDefinition(name, opts...)

50 > globalRegistry.register(def)

51 > return counterDefinition{def}

52 > }

53

54 > func NewGaugeDef(name string, opts ...Option) gaugeDefinition { defs.go

55 > // This line cannot be combined with others!

56 > // This ensures the stack trace has information of the caller.

57 > def := newMetricDefinition(name, opts...)

58 > globalRegistry.register(def)

59 > return gaugeDefinition{def}

60 > }

61

62 func (d histogramDefinition) With(handler Handler) HistogramIface {

68 }

69

70 > func (d gaugeDefinition) With(handler Handler) GaugeIface { defs.go

71 > return handler.Gauge(d.name)

72 > }

73

74 > func (d timerDefinition) With(handler Handler) TimerIface { defs.go

75 > return handler.Timer(d.name)

76 > }

go.temporal.io/server/api/adminservice/v1/request_response.pb.go 36 covered LOC · 1 range

Open complete file

6623 }

6624

6625 > func init() { file_temporal_server_api_adminservice_v1_request_response_proto_init() } request_response.pb.go

6626 > func file_temporal_server_api_adminservice_v1_request_response_proto_init() {

6627 > if File_temporal_server_api_adminservice_v1_request_response_proto != nil {

6628 > return

6629 > }

6630 > file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[59].OneofWrappers = []any{

6631 > (*StreamWorkflowReplicationMessagesRequest_SyncReplicationState)(nil),

6632 > }

6633 > file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[60].OneofWrappers = []any{

6634 > (*StreamWorkflowReplicationMessagesResponse_Messages)(nil),

6635 > }

6636 > file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[61].OneofWrappers = []any{

6637 > (*GetNamespaceRequest_Namespace)(nil),

6638 > (*GetNamespaceRequest_Id)(nil),

6639 > }

6640 > file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes[90].OneofWrappers = []any{

6641 > (*StartAdminBatchOperationRequest_RefreshTasksOperation)(nil),

6642 > }

6643 > type x struct{}

6644 > out := protoimpl.TypeBuilder{

6645 > File: protoimpl.DescBuilder{

6646 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

6647 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_adminservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_adminservice_v1_request_response_proto_rawDesc)),

6648 > NumEnums: 1,

6649 > NumMessages: 105,

6650 > NumExtensions: 0,

6651 > NumServices: 0,

6652 > },

6653 > GoTypes: file_temporal_server_api_adminservice_v1_request_response_proto_goTypes,

6654 > DependencyIndexes: file_temporal_server_api_adminservice_v1_request_response_proto_depIdxs,

6655 > EnumInfos: file_temporal_server_api_adminservice_v1_request_response_proto_enumTypes,

6656 > MessageInfos: file_temporal_server_api_adminservice_v1_request_response_proto_msgTypes,

6657 > }.Build()

6658 > File_temporal_server_api_adminservice_v1_request_response_proto = out.File

6659 > file_temporal_server_api_adminservice_v1_request_response_proto_goTypes = nil

6660 > file_temporal_server_api_adminservice_v1_request_response_proto_depIdxs = nil

6661 }

go.temporal.io/server/api/replication/v1/message.pb.go 36 covered LOC · 2 ranges

Open complete file

2441 }

2442

2443 > func init() { file_temporal_server_api_replication_v1_message_proto_init() } message.pb.go

2444 > func file_temporal_server_api_replication_v1_message_proto_init() {

2445 > if File_temporal_server_api_replication_v1_message_proto != nil {

2446 return

2447 }

2448 > file_temporal_server_api_replication_v1_message_proto_msgTypes[0].OneofWrappers = []any{ message.pb.go

2449 > (*ReplicationTask_NamespaceTaskAttributes)(nil),

2450 > (*ReplicationTask_SyncShardStatusTaskAttributes)(nil),

2451 > (*ReplicationTask_SyncActivityTaskAttributes)(nil),

2452 > (*ReplicationTask_HistoryTaskAttributes)(nil),

2453 > (*ReplicationTask_SyncWorkflowStateTaskAttributes)(nil),

2454 > (*ReplicationTask_TaskQueueUserDataAttributes)(nil),

2455 > (*ReplicationTask_SyncHsmAttributes)(nil),

2456 > (*ReplicationTask_BackfillHistoryTaskAttributes)(nil),

2457 > (*ReplicationTask_VerifyVersionedTransitionTaskAttributes)(nil),

2458 > (*ReplicationTask_SyncVersionedTransitionTaskAttributes)(nil),

2459 > }

2460 > file_temporal_server_api_replication_v1_message_proto_msgTypes[21].OneofWrappers = []any{

2461 > (*VersionedTransitionArtifact_SyncWorkflowStateMutationAttributes)(nil),

2462 > (*VersionedTransitionArtifact_SyncWorkflowStateSnapshotAttributes)(nil),

2463 > }

2464 > type x struct{}

2465 > out := protoimpl.TypeBuilder{

2466 > File: protoimpl.DescBuilder{

2467 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

2468 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_replication_v1_message_proto_rawDesc), len(file_temporal_server_api_replication_v1_message_proto_rawDesc)),

2469 > NumEnums: 0,

2470 > NumMessages: 23,

2471 > NumExtensions: 0,

2472 > NumServices: 0,

2473 > },

2474 > GoTypes: file_temporal_server_api_replication_v1_message_proto_goTypes,

2475 > DependencyIndexes: file_temporal_server_api_replication_v1_message_proto_depIdxs,

2476 > MessageInfos: file_temporal_server_api_replication_v1_message_proto_msgTypes,

2477 > }.Build()

2478 > File_temporal_server_api_replication_v1_message_proto = out.File

2479 > file_temporal_server_api_replication_v1_message_proto_goTypes = nil

2480 > file_temporal_server_api_replication_v1_message_proto_depIdxs = nil

2481 }

go.temporal.io/server/common/backoff/retrypolicy.go 34 covered LOC · 6 ranges

Open complete file

80

81 // NewExponentialRetryPolicy returns an instance of ExponentialRetryPolicy using the provided initialInterval

82 > func NewExponentialRetryPolicy(initialInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go

83 > p := &ExponentialRetryPolicy{

84 > initialInterval: initialInterval,

85 > backoffCoefficient: defaultBackoffCoefficient,

86 > maximumInterval: defaultMaximumInterval,

87 > expirationInterval: defaultExpirationInterval,

88 > maximumAttempts: defaultMaximumAttempts,

89 > }

90 >

91 > return p

92 > }

93

94 // NewRetrier is used for creating a new instance of Retrier

121 // This does *not* cause the policy to stop retrying when the interval between retries reaches the supplied duration.

122 // That is what WithExpirationInterval does. Instead, this prevents the interval from exceeding maximumInterval.

123 > func (p *ExponentialRetryPolicy) WithMaximumInterval(maximumInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go

124 > p.maximumInterval = maximumInterval

125 > return p

126 > }

127

128 // WithExpirationInterval sets the absolute expiration interval for all retries

129 > func (p *ExponentialRetryPolicy) WithExpirationInterval(expirationInterval time.Duration) *ExponentialRetryPolicy { retrypolicy.go

130 > p.expirationInterval = expirationInterval

131 > return p

132 > }

133

134 // WithMaximumAttempts sets the maximum number of retry attempts

135 > func (p *ExponentialRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ExponentialRetryPolicy { retrypolicy.go

136 > p.maximumAttempts = maximumAttempts

137 > return p

138 > }

139

140 // ComputeNextDelay returns the next delay interval. This is used by Retrier to delay calling the operation again

267 var _ RetryPolicy = (*ConstantDelayRetryPolicy)(nil)

268

269 > func NewConstantDelayRetryPolicy(delay time.Duration) *ConstantDelayRetryPolicy { retrypolicy.go

270 > return &ConstantDelayRetryPolicy{

271 > maximumAttempts: defaultMaximumAttempts,

272 > jitterPct: defaultJitterPct,

273 > delay: delay,

274 > }

275 > }

276

277 > func (p *ConstantDelayRetryPolicy) WithMaximumAttempts(maximumAttempts int) *ConstantDelayRetryPolicy { retrypolicy.go

278 > p.maximumAttempts = maximumAttempts

279 > return p

280 > }

281

282 func (p *ConstantDelayRetryPolicy) WithJitter(jitterPct float64) *ConstantDelayRetryPolicy {

go.temporal.io/server/api/persistence/v1/chasm.pb.go 33 covered LOC · 1 range

Open complete file

1180 }

1181

1182 > func init() { file_temporal_server_api_persistence_v1_chasm_proto_init() } chasm.pb.go

1183 > func file_temporal_server_api_persistence_v1_chasm_proto_init() {

1184 > if File_temporal_server_api_persistence_v1_chasm_proto != nil {

1185 > return

1186 > }

1187 > file_temporal_server_api_persistence_v1_hsm_proto_init()

1188 > file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[1].OneofWrappers = []any{

1189 > (*ChasmNodeMetadata_ComponentAttributes)(nil),

1190 > (*ChasmNodeMetadata_DataAttributes)(nil),

1191 > (*ChasmNodeMetadata_CollectionAttributes)(nil),

1192 > (*ChasmNodeMetadata_PointerAttributes)(nil),

1193 > }

1194 > file_temporal_server_api_persistence_v1_chasm_proto_msgTypes[10].OneofWrappers = []any{

1195 > (*ChasmNexusCompletion_Success)(nil),

1196 > (*ChasmNexusCompletion_Failure)(nil),

1197 > }

1198 > type x struct{}

1199 > out := protoimpl.TypeBuilder{

1200 > File: protoimpl.DescBuilder{

1201 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

1202 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_proto_rawDesc)),

1203 > NumEnums: 0,

1204 > NumMessages: 15,

1205 > NumExtensions: 0,

1206 > NumServices: 0,

1207 > },

1208 > GoTypes: file_temporal_server_api_persistence_v1_chasm_proto_goTypes,

1209 > DependencyIndexes: file_temporal_server_api_persistence_v1_chasm_proto_depIdxs,

1210 > MessageInfos: file_temporal_server_api_persistence_v1_chasm_proto_msgTypes,

1211 > }.Build()

1212 > File_temporal_server_api_persistence_v1_chasm_proto = out.File

1213 > file_temporal_server_api_persistence_v1_chasm_proto_goTypes = nil

1214 > file_temporal_server_api_persistence_v1_chasm_proto_depIdxs = nil

1215 }

go.temporal.io/server/service/worker/scheduler/buffer.go 33 covered LOC · 10 ranges

Open complete file

32 isRunning bool,

33 resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy,

34 > ) ProcessBufferResult[T] { buffer.go

35 > // We should try to do something reasonable with any combination of overlap

36 > // policies in the buffer, although some combinations don't make much sense

37 > // and would require a convoluted series of calls to set up.

38 > //

39 > // Buffer entries that have unspecified overlap policy are resolved to the

40 > // current policy here, not earlier, so that updates to the policy can

41 > // affect them.

42 >

43 > var action ProcessBufferResult[T]

44 > action.OverlapSkippedByPolicy = make(map[enumspb.ScheduleOverlapPolicy]int64)

45 > var zeroVal T

46 >

47 > for _, start := range buffer {

48 > overlapPolicy := resolve(start.GetOverlapPolicy()) buffer.go

49 >

50 > // For allow-all, just collect and start all at once

51 > if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL {

52 > action.OverlappingStarts = append(action.OverlappingStarts, start) buffer.go

53 > continue

54 }

55

57

58 // If there's nothing running, we can start this one no matter what the policy is

59 > if !isRunning && action.NonOverlappingStart == zeroVal { buffer.go

60 > action.NonOverlappingStart = start buffer.go

61 > continue

62 }

63

64 // Otherwise this one overlaps and we should apply the policy

65 > switch overlapPolicy { buffer.go

66 > case enumspb.SCHEDULE_OVERLAP_POLICY_SKIP: buffer.go

67 > // just skip

68 > action.OverlapSkipped++

69 > action.OverlapSkippedByPolicy[overlapPolicy]++

70 case enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE:

71 // allow one (the first one) in the buffer

76 action.OverlapSkippedByPolicy[overlapPolicy]++

77 }

78 > case enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL: buffer.go

79 > // always add to buffer

80 > action.NewBuffer = append(action.NewBuffer, start)

81 case enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER:

82 if isRunning {

102 }

103

104 > if action.NeedCancel || action.NeedTerminate { buffer.go

105 // In a very contrived situation, we could end up with overlapping starts at the same

106 // time as a non-overlapping start tries to cancel/terminate a running workflow. But

110 }

111

112 > return action buffer.go

113 }

go.temporal.io/server/api/persistence/v1/hsm.pb.go 32 covered LOC · 1 range

Open complete file

895 }

896

897 > func init() { file_temporal_server_api_persistence_v1_hsm_proto_init() } hsm.pb.go

898 > func file_temporal_server_api_persistence_v1_hsm_proto_init() {

899 > if File_temporal_server_api_persistence_v1_hsm_proto != nil {

900 > return

901 > }

902 > file_temporal_server_api_persistence_v1_hsm_proto_msgTypes[8].OneofWrappers = []any{

903 > (*StateMachineTombstone_ActivityScheduledEventId)(nil),

904 > (*StateMachineTombstone_TimerId)(nil),

905 > (*StateMachineTombstone_ChildExecutionInitiatedEventId)(nil),

906 > (*StateMachineTombstone_RequestCancelInitiatedEventId)(nil),

907 > (*StateMachineTombstone_SignalExternalInitiatedEventId)(nil),

908 > (*StateMachineTombstone_UpdateId)(nil),

909 > (*StateMachineTombstone_StateMachinePath)(nil),

910 > (*StateMachineTombstone_ChasmNodePath)(nil),

911 > }

912 > type x struct{}

913 > out := protoimpl.TypeBuilder{

914 > File: protoimpl.DescBuilder{

915 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

916 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_hsm_proto_rawDesc), len(file_temporal_server_api_persistence_v1_hsm_proto_rawDesc)),

917 > NumEnums: 0,

918 > NumMessages: 12,

919 > NumExtensions: 0,

920 > NumServices: 0,

921 > },

922 > GoTypes: file_temporal_server_api_persistence_v1_hsm_proto_goTypes,

923 > DependencyIndexes: file_temporal_server_api_persistence_v1_hsm_proto_depIdxs,

924 > MessageInfos: file_temporal_server_api_persistence_v1_hsm_proto_msgTypes,

925 > }.Build()

926 > File_temporal_server_api_persistence_v1_hsm_proto = out.File

927 > file_temporal_server_api_persistence_v1_hsm_proto_goTypes = nil

928 > file_temporal_server_api_persistence_v1_hsm_proto_depIdxs = nil

929 }

go.temporal.io/server/api/persistence/v1/update.pb.go 31 covered LOC · 1 range

Open complete file

422 }

423

424 > func init() { file_temporal_server_api_persistence_v1_update_proto_init() } update.pb.go

425 > func file_temporal_server_api_persistence_v1_update_proto_init() {

426 > if File_temporal_server_api_persistence_v1_update_proto != nil {

427 > return

428 > }

429 > file_temporal_server_api_persistence_v1_hsm_proto_init()

430 > file_temporal_server_api_persistence_v1_update_proto_msgTypes[0].OneofWrappers = []any{

431 > (*UpdateAdmissionInfo_HistoryPointer_)(nil),

432 > }

433 > file_temporal_server_api_persistence_v1_update_proto_msgTypes[3].OneofWrappers = []any{

434 > (*UpdateInfo_Acceptance)(nil),

435 > (*UpdateInfo_Completion)(nil),

436 > (*UpdateInfo_Admission)(nil),

437 > }

438 > type x struct{}

439 > out := protoimpl.TypeBuilder{

440 > File: protoimpl.DescBuilder{

441 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

442 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_update_proto_rawDesc), len(file_temporal_server_api_persistence_v1_update_proto_rawDesc)),

443 > NumEnums: 0,

444 > NumMessages: 5,

445 > NumExtensions: 0,

446 > NumServices: 0,

447 > },

448 > GoTypes: file_temporal_server_api_persistence_v1_update_proto_goTypes,

449 > DependencyIndexes: file_temporal_server_api_persistence_v1_update_proto_depIdxs,

450 > MessageInfos: file_temporal_server_api_persistence_v1_update_proto_msgTypes,

451 > }.Build()

452 > File_temporal_server_api_persistence_v1_update_proto = out.File

453 > file_temporal_server_api_persistence_v1_update_proto_goTypes = nil

454 > file_temporal_server_api_persistence_v1_update_proto_depIdxs = nil

455 }

go.temporal.io/server/common/dynamicconfig/deepcopy.go 31 covered LOC · 6 ranges

Open complete file

9 // deepCopyForMapstructure does a simple deep copy of T. Fancy cases (anything other than plain old data)

10 // is not handled and will panic.

11 > func deepCopyForMapstructure[T any](t T) T { deepcopy.go

12 > // nolint:revive // this will be triggered from a static initializer before it can be triggered from production code

13 > return deepCopyValue(reflect.ValueOf(t)).Interface().(T)

14 > }

15

16 > func deepCopyValue(v reflect.Value) reflect.Value { deepcopy.go

17 > switch v.Kind() {

18 case reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64,

19 reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64,

20 > reflect.Uintptr, reflect.Float32, reflect.Float64, reflect.String: deepcopy.go

21 > nv := reflect.New(v.Type()).Elem()

22 > nv.Set(v)

23 > return nv

24 case reflect.Array:

25 nv := reflect.New(v.Type()).Elem()

42 }

43 return deepCopyValue(v.Elem()).Addr()

44 > case reflect.Slice: deepcopy.go

45 > if v.IsNil() {

46 > return v

47 > }

48 nv := reflect.MakeSlice(v.Type(), v.Len(), v.Len())

49 for i := range v.Len() {

51 }

52 return nv

53 > case reflect.Struct: deepcopy.go

54 > // Special case for time.Time: it has unexported fields so we can't copy it field by

55 > // field, but we can copy zero values (which is all we need for default values).

56 > if v.Type() == reflect.TypeFor[time.Time]() {

57 > if v.Interface().(time.Time).IsZero() {

58 > return reflect.ValueOf(time.Time{})

59 > }

60 // nolint:forbidigo // this will be triggered from a static initializer before it can be triggered from production code

61 panic(fmt.Sprintf("Can't deep copy non-zero time.Time: %v", v.Interface()))

62 }

63 > nv := reflect.New(v.Type()).Elem() deepcopy.go

64 > for i := range v.Type().NumField() {

65 > nv.Field(i).Set(deepCopyValue(v.Field(i)))

66 > }

67 > return nv

68 > case reflect.Interface, reflect.Func, reflect.Chan:

69 > // only nil values of any other reference types allowed!

70 > if v.IsNil() {

71 > return v

72 > }

73 fallthrough

74 default:

go.temporal.io/server/common/log/sdk_logger.go 31 covered LOC · 14 ranges

Open complete file

19 var _ log.Logger = (*SdkLogger)(nil)

20

21 > func NewSdkLogger(logger Logger) *SdkLogger { sdk_logger.go

22 > if sl, ok := logger.(SkipLogger); ok {

23 > logger = sl.Skip(extraSkipForSdkLogger) sdk_logger.go

24 > }

25

26 > return &SdkLogger{ sdk_logger.go

27 > logger: logger,

28 > }

29 }

30

31 > func (l *SdkLogger) tags(keyvals []any) []tag.Tag { sdk_logger.go

32 > var tags []tag.Tag

33 > for i := 0; i < len(keyvals); i++ {

34 > if t, keyvalIsTag := keyvals[i].(tag.Tag); keyvalIsTag { sdk_logger.go

35 tags = append(tags, t)

36 continue

37 }

38

39 > key, keyIsString := keyvals[i].(string) sdk_logger.go

40 > if !keyIsString {

41 key = fmt.Sprintf("%v", keyvals[i])

42 }

43 > var val any sdk_logger.go

44 > if i+1 == len(keyvals) {

45 val = noValue

46 > } else { sdk_logger.go

47 > val = keyvals[i+1] sdk_logger.go

48 > i++

49 > }

50

51 > tags = append(tags, tag.Any(key, val)) sdk_logger.go

52 }

53

54 > return tags sdk_logger.go

55 }

56

57 > func (l *SdkLogger) Debug(msg string, keyvals ...any) { sdk_logger.go

58 > l.logger.Debug(msg, l.tags(keyvals)...)

59 > }

60

61 > func (l *SdkLogger) Info(msg string, keyvals ...any) { sdk_logger.go

62 > l.logger.Info(msg, l.tags(keyvals)...)

63 > }

64

65 func (l *SdkLogger) Warn(msg string, keyvals ...any) {

71 }

72

73 > func (l *SdkLogger) With(keyvals ...any) log.Logger { sdk_logger.go

74 > return NewSdkLogger(

75 > With(l.logger, l.tags(keyvals)...))

76 > }

go.temporal.io/server/common/persistence/sql/sqlplugin/mysql/execution_maps.go 30 covered LOC · 5 ranges

Open complete file

53 )

54

55 > func stringMap(a []string, f func(string) string) []string { execution_maps.go

56 > b := make([]string, len(a))

57 > for i, v := range a {

58 > b[i] = f(v)

59 > }

60 > return b

61 }

62

63 > func makeDeleteMapQry(tableName string) string { execution_maps.go

64 > return fmt.Sprintf(deleteMapQryTemplate, tableName)

65 > }

66

67 > func makeSetKeyInMapQry(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { execution_maps.go

68 > return fmt.Sprintf(setKeyInMapQryTemplate,

69 > tableName,

70 > strings.Join(nonPrimaryKeyColumns, ","),

71 > strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {

72 > return ":" + x

73 > }), ","),

74 > strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {

75 > return x + "=VALUES(" + x + ")"

76 > }), ","),

77 mapKeyName)

78 }

79

80 > func makeDeleteKeyInMapQry(tableName string, mapKeyName string) string { execution_maps.go

81 > return fmt.Sprintf(deleteKeyInMapQryTemplate,

82 > tableName,

83 > mapKeyName)

84 > }

85

86 > func makeGetMapQryTemplate(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { execution_maps.go

87 > return fmt.Sprintf(getMapQryTemplate,

88 > tableName,

89 > mapKeyName,

90 > strings.Join(nonPrimaryKeyColumns, ","))

91 > }

92

93 var (

go.temporal.io/server/common/persistence/sql/sqlplugin/postgresql/execution_maps.go 30 covered LOC · 6 ranges

Open complete file

86 )

87

88 > func stringMap(a []string, f func(string) string) []string { execution_maps.go

89 > b := make([]string, len(a))

90 > for i, v := range a {

91 > b[i] = f(v)

92 > }

93 > return b

94 }

95

96 > func makeDeleteMapQry(tableName string) string { execution_maps.go

97 > return fmt.Sprintf(deleteMapQueryTemplate, tableName)

98 > }

99

100 > func makeSetKeyInMapQry(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { execution_maps.go

101 > return fmt.Sprintf(setKeyInMapQueryTemplate,

102 > tableName,

103 > strings.Join(nonPrimaryKeyColumns, ","),

104 > strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {

105 > return ":" + x

106 > }), ","),

107 mapKeyName,

108 > strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string { execution_maps.go

109 > return "excluded." + x

110 > }), ","))

111 }

112

113 > func makeDeleteKeyInMapQry(tableName string, mapKeyName string) string { execution_maps.go

114 > return fmt.Sprintf(deleteKeyInMapQueryTemplate,

115 > tableName,

116 > mapKeyName)

117 > }

118

119 > func makeGetMapQryTemplate(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { execution_maps.go

120 > return fmt.Sprintf(getMapQueryTemplate,

121 > tableName,

122 > mapKeyName,

123 > strings.Join(nonPrimaryKeyColumns, ","))

124 > }

125

126 var (

go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite/execution_maps.go 30 covered LOC · 5 ranges

Open complete file

52 )

53

54 > func stringMap(a []string, f func(string) string) []string { execution_maps.go

55 > b := make([]string, len(a))

56 > for i, v := range a {

57 > b[i] = f(v)

58 > }

59 > return b

60 }

61

62 > func makeDeleteMapQry(tableName string) string { execution_maps.go

63 > return fmt.Sprintf(deleteMapQryTemplate, tableName)

64 > }

65

66 > func makeSetKeyInMapQry(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { execution_maps.go

67 > return fmt.Sprintf(setKeyInMapQryTemplate,

68 > tableName,

69 > strings.Join(nonPrimaryKeyColumns, ","),

70 > strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {

71 > return ":" + x

72 > }), ","),

73 > strings.Join(stringMap(nonPrimaryKeyColumns, func(x string) string {

74 > return x + "=" + x

75 > }), ","),

76 mapKeyName)

77 }

78

79 > func makeDeleteKeyInMapQry(tableName string, mapKeyName string) string { execution_maps.go

80 > return fmt.Sprintf(deleteKeyInMapQryTemplate,

81 > tableName,

82 > mapKeyName)

83 > }

84

85 > func makeGetMapQryTemplate(tableName string, nonPrimaryKeyColumns []string, mapKeyName string) string { execution_maps.go

86 > return fmt.Sprintf(getMapQryTemplate,

87 > tableName,

88 > mapKeyName,

89 > strings.Join(nonPrimaryKeyColumns, ","))

90 > }

91

92 var (

go.temporal.io/server/common/searchattribute/sadefs/constants.go 30 covered LOC · 9 ranges

Open complete file

261 }

262

263 > dbCustomSearchAttributeFieldNameRE = func() map[enumspb.IndexedValueType]*regexp.Regexp { constants.go

264 > res := map[enumspb.IndexedValueType]*regexp.Regexp{}

265 > for t := range defaultNumDBCustomSearchAttributes {

266 > res[t] = regexp.MustCompile(fmt.Sprintf(`^%s(0[1-9]|[1-9][0-9])$`, t.String()))

267 > }

268 > return res

269 }()

270 )

271

272 // System returns a clone of the system search attributes map.

273 > func System() map[string]enumspb.IndexedValueType { constants.go

274 > return maps.Clone(system)

275 > }

276

277 // Predefined returns a clone of the predefined search attributes map.

278 > func Predefined() map[string]enumspb.IndexedValueType { constants.go

279 > return maps.Clone(predefined)

280 > }

281

282 // PredefinedWhiteList returns a clone of the predefined whitelist search attributes map.

283 > func PredefinedWhiteList() map[string]enumspb.IndexedValueType { constants.go

284 > return maps.Clone(predefinedWhiteList)

285 > }

286

287 // Reserved returns a clone of the reserved field names map.

343 // GetSqlDbColName maps system and reserved search attributes to column names for SQL tables.

344 // If the input is not a system or reserved search attribute, then it returns the input.

345 > func GetSqlDbColName(name string) string { constants.go

346 > if fieldName, ok := sqlDbSystemNameToColName[name]; ok {

347 > return fieldName constants.go

348 > }

349 return name

350 }

352 func GetDBIndexSearchAttributes(

353 override map[enumspb.IndexedValueType]int,

354 > ) *persistencespb.IndexSearchAttributes { constants.go

355 > csa := map[string]enumspb.IndexedValueType{}

356 > for saType, defaultNumAttrs := range defaultNumDBCustomSearchAttributes {

357 > numAttrs := defaultNumAttrs

358 > if value, ok := override[saType]; ok {

359 numAttrs = value

360 }

361 > for i := range numAttrs { constants.go

362 > csa[fmt.Sprintf("%s%02d", saType.String(), i+1)] = saType

363 > }

364 }

365 > return &persistencespb.IndexSearchAttributes{ constants.go

366 > CustomSearchAttributes: csa,

367 > }

368 }

369

go.temporal.io/server/api/taskqueue/v1/message.pb.go 29 covered LOC · 2 ranges

Open complete file

1330 }

1331

1332 > func init() { file_temporal_server_api_taskqueue_v1_message_proto_init() } message.pb.go

1333 > func file_temporal_server_api_taskqueue_v1_message_proto_init() {

1334 > if File_temporal_server_api_taskqueue_v1_message_proto != nil {

1335 return

1336 }

1337 > file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[0].OneofWrappers = []any{ message.pb.go

1338 > (*TaskVersionDirective_UseAssignmentRules)(nil),

1339 > (*TaskVersionDirective_AssignedBuildId)(nil),

1340 > }

1341 > file_temporal_server_api_taskqueue_v1_message_proto_msgTypes[5].OneofWrappers = []any{

1342 > (*TaskQueuePartition_NormalPartitionId)(nil),

1343 > (*TaskQueuePartition_StickyName)(nil),

1344 > (*TaskQueuePartition_WorkerCommands)(nil),

1345 > }

1346 > type x struct{}

1347 > out := protoimpl.TypeBuilder{

1348 > File: protoimpl.DescBuilder{

1349 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

1350 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_taskqueue_v1_message_proto_rawDesc), len(file_temporal_server_api_taskqueue_v1_message_proto_rawDesc)),

1351 > NumEnums: 0,

1352 > NumMessages: 16,

1353 > NumExtensions: 0,

1354 > NumServices: 0,

1355 > },

1356 > GoTypes: file_temporal_server_api_taskqueue_v1_message_proto_goTypes,

1357 > DependencyIndexes: file_temporal_server_api_taskqueue_v1_message_proto_depIdxs,

1358 > MessageInfos: file_temporal_server_api_taskqueue_v1_message_proto_msgTypes,

1359 > }.Build()

1360 > File_temporal_server_api_taskqueue_v1_message_proto = out.File

1361 > file_temporal_server_api_taskqueue_v1_message_proto_goTypes = nil

1362 > file_temporal_server_api_taskqueue_v1_message_proto_depIdxs = nil

1363 }

go.temporal.io/server/common/log/tag/tags.go 27 covered LOC · 9 ranges

Open complete file

70

71 // WorkflowAction returns tag for WorkflowAction

72 > func workflowAction(action string) ZapTag { tags.go

73 > return NewStringTag("wf-action", action)

74 > }

75

76 // WorkflowListFilterType returns tag for WorkflowListFilterType

77 > func workflowListFilterType(listFilterType string) ZapTag { tags.go

78 > return NewStringTag("wf-list-filter-type", listFilterType)

79 > }

80

81 // general

376

377 // Component returns tag for Component

378 > func component(component string) ZapTag { tags.go

379 > return NewStringTag("component", component)

380 > }

381

382 // Lifecycle returns tag for Lifecycle

383 > func lifecycle(lifecycle string) ZapTag { tags.go

384 > return NewStringTag("lifecycle", lifecycle)

385 > }

386

387 // StoreOperation returns tag for StoreOperation

388 > func storeOperation(storeOperation string) ZapTag { tags.go

389 > return NewStringTag("store-operation", storeOperation)

390 > }

391

392 // OperationResult returns tag for OperationResult

393 > func operationResult(operationResult string) ZapTag { tags.go

394 > return NewStringTag("operation-result", operationResult)

395 > }

396

397 // ErrorType returns tag for ErrorType

401

402 // errorType returns tag for ErrorType given a string

403 > func errorType(errorType string) ZapTag { tags.go

404 > return NewStringTag("error-type", errorType)

405 > }

406

407 // Shardupdate returns tag for Shardupdate

408 > func shardupdate(shardupdate string) ZapTag { tags.go

409 > return NewStringTag("shard-update", shardupdate)

410 > }

411

412 // scope returns a tag for scope

413 // Pre-defined scope tags are in values.go.

414 > func scope(scope string) ZapTag { tags.go

415 > return NewStringTag("scope", scope)

416 > }

417

418 // general

go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1/message.pb.go 26 covered LOC · 1 range

Open complete file

843 }

844

845 > func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_init() } message.pb.go

846 > func file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_init() {

847 > if File_temporal_server_chasm_lib_scheduler_proto_v1_message_proto != nil {

848 > return

849 > }

850 > file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_msgTypes[4].OneofWrappers = []any{

851 > (*BackfillerState_BackfillRequest)(nil),

852 > (*BackfillerState_TriggerRequest)(nil),

853 > }

854 > type x struct{}

855 > out := protoimpl.TypeBuilder{

856 > File: protoimpl.DescBuilder{

857 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

858 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_rawDesc)),

859 > NumEnums: 0,

860 > NumMessages: 12,

861 > NumExtensions: 0,

862 > NumServices: 0,

863 > },

864 > GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_goTypes,

865 > DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_depIdxs,

866 > MessageInfos: file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_msgTypes,

867 > }.Build()

868 > File_temporal_server_chasm_lib_scheduler_proto_v1_message_proto = out.File

869 > file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_goTypes = nil

870 > file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_depIdxs = nil

871 }

go.temporal.io/server/common/util/util.go 26 covered LOC · 14 ranges

Open complete file

12

13 // MinTime returns the earlier of two given time.Time

14 > func MinTime(a, b time.Time) time.Time { util.go

15 > if a.Before(b) {

16 > return a util.go

17 > }

18 > return b util.go

19 }

20

21 // MaxTime returns the latest of the given time.Time values.

22 > func MaxTime(first time.Time, rest ...time.Time) time.Time { util.go

23 > latest := first

24 > for _, t := range rest {

25 > if t.After(latest) {

26 > latest = t util.go

27 > }

28 }

29 > return latest util.go

30 }

31

43

44 // SliceHead returns the first n elements of s. n may be greater than len(s).

45 > func SliceHead[S ~[]E, E any](s S, n int) S { util.go

46 > if n < len(s) {

47 return s[:n]

48 }

49 > return s util.go

50 }

51

52 // SliceTail returns the last n elements of s. n may be greater than len(s).

53 > func SliceTail[S ~[]E, E any](s S, n int) S { util.go

54 > if extra := len(s) - n; extra > 0 {

55 > return s[extra:] util.go

56 > }

57 > return s util.go

58 }

59

60 // CloneMapNonNil is like maps.Clone except it can't return nil, it will return an empty map instead.

61 > func CloneMapNonNil[M ~map[K]V, K comparable, V any](m M) M { util.go

62 > m = maps.Clone(m)

63 > if m == nil {

64 > m = make(M) util.go

65 > }

66 > return m util.go

67 }

68

go.temporal.io/server/common/routing/route.go 25 covered LOC · 8 ranges

Open complete file

36

37 // NewRoute returns a new [Route] instance with the given components.

38 > func NewRoute[T any](components ...Component[T]) Route[T] { route.go

39 > return Route[T]{components: components}

40 > }

41

42 // RouteBuilder is a builder for the [Route] interface.

46

47 // NewBuilder creates a new [RouteBuilder] instance, which can be used to define a new [Route] via a fluent API.

48 > func NewBuilder[T any]() *RouteBuilder[T] { route.go

49 > return &RouteBuilder[T]{}

50 > }

51

52 // With adds a series of [Component] instances to the [Route].

53 > func (r *RouteBuilder[T]) With(c ...Component[T]) *RouteBuilder[T] { route.go

54 > r.components = append(r.components, c...)

55 > return r

56 > }

57

58 // Constant adds a [Constant] component to the [Route].

59 > func (r *RouteBuilder[T]) Constant(values ...string) *RouteBuilder[T] { route.go

60 > return r.With(Constant[T](values...))

61 > }

62

63 // StringVariable adds a [StringVariable] component to the [Route].

64 > func (r *RouteBuilder[T]) StringVariable(name string, getter func(*T) *string) *RouteBuilder[T] { route.go

65 > return r.With(StringVariable[T](name, getter))

66 > }

67

68 // Build returns a read-only [Route].

69 > func (r *RouteBuilder[T]) Build() Route[T] { route.go

70 > return NewRoute[T](r.components...)

71 > }

72

73 // Representation returns the [github.com/gorilla/mux] compatible string representation of the route for usage in a

111 // Constant returns a [Component] that represents a series of constant HTTP path components in a Route.

112 // They will be joined via strings when used to construct a path or path representation.

113 > func Constant[T any](values ...string) constant[T] { route.go

114 > return values

115 > }

116

117 type constant[T any] []string

128

129 // StringVariable returns a [Component] that represents a string variable in a Route.

130 > func StringVariable[T any](name string, getter func(*T) *string) stringVariable[T] { route.go

131 > return stringVariable[T]{name, getter}

132 > }

133

134 type stringVariable[T any] struct {

go.temporal.io/server/api/persistence/v1/nexus.pb.go 24 covered LOC · 2 ranges

Open complete file

481 }

482

483 > func init() { file_temporal_server_api_persistence_v1_nexus_proto_init() } nexus.pb.go

484 > func file_temporal_server_api_persistence_v1_nexus_proto_init() {

485 > if File_temporal_server_api_persistence_v1_nexus_proto != nil {

486 return

487 }

488 > file_temporal_server_api_persistence_v1_nexus_proto_msgTypes[1].OneofWrappers = []any{ nexus.pb.go

489 > (*NexusEndpointTarget_Worker_)(nil),

490 > (*NexusEndpointTarget_External_)(nil),

491 > }

492 > type x struct{}

493 > out := protoimpl.TypeBuilder{

494 > File: protoimpl.DescBuilder{

495 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

496 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_nexus_proto_rawDesc), len(file_temporal_server_api_persistence_v1_nexus_proto_rawDesc)),

497 > NumEnums: 0,

498 > NumMessages: 6,

499 > NumExtensions: 0,

500 > NumServices: 0,

501 > },

502 > GoTypes: file_temporal_server_api_persistence_v1_nexus_proto_goTypes,

503 > DependencyIndexes: file_temporal_server_api_persistence_v1_nexus_proto_depIdxs,

504 > MessageInfos: file_temporal_server_api_persistence_v1_nexus_proto_msgTypes,

505 > }.Build()

506 > File_temporal_server_api_persistence_v1_nexus_proto = out.File

507 > file_temporal_server_api_persistence_v1_nexus_proto_goTypes = nil

508 > file_temporal_server_api_persistence_v1_nexus_proto_depIdxs = nil

509 }

go.temporal.io/server/api/persistence/v1/workflow_mutable_state.pb.go 24 covered LOC · 2 ranges

Open complete file

532 }

533

534 > func init() { file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_init() } workflow_mutable_state.pb.go

535 > func file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_init() {

536 > if File_temporal_server_api_persistence_v1_workflow_mutable_state_proto != nil {

537 return

538 }

539 > file_temporal_server_api_persistence_v1_chasm_proto_init() workflow_mutable_state.pb.go

540 > file_temporal_server_api_persistence_v1_executions_proto_init()

541 > file_temporal_server_api_persistence_v1_hsm_proto_init()

542 > file_temporal_server_api_persistence_v1_update_proto_init()

543 > type x struct{}

544 > out := protoimpl.TypeBuilder{

545 > File: protoimpl.DescBuilder{

546 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

547 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_rawDesc), len(file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_rawDesc)),

548 > NumEnums: 0,

549 > NumMessages: 16,

550 > NumExtensions: 0,

551 > NumServices: 0,

552 > },

553 > GoTypes: file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_goTypes,

554 > DependencyIndexes: file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_depIdxs,

555 > MessageInfos: file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_msgTypes,

556 > }.Build()

557 > File_temporal_server_api_persistence_v1_workflow_mutable_state_proto = out.File

558 > file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_goTypes = nil

559 > file_temporal_server_api_persistence_v1_workflow_mutable_state_proto_depIdxs = nil

560 }

go.temporal.io/server/common/log/tag/zap_tag.go 24 covered LOC · 6 ranges

Open complete file

26 }

27

28 > func (t ZapTag) Field() zap.Field { zap_tag.go

29 > return t.field

30 > }

31

32 > func (t ZapTag) Key() string { zap_tag.go

33 > return t.field.Key

34 > }

35

36 func (t ZapTag) Value() any {

44 }

45

46 > func NewStringTag(key string, value string) ZapTag { zap_tag.go

47 > return ZapTag{

48 > field: zap.String(key, value),

49 > }

50 > }

51

52 func NewStringsTag(key string, value []string) ZapTag {

118 }

119

120 > func NewBoolTag(key string, value bool) ZapTag { zap_tag.go

121 > return ZapTag{

122 > field: zap.Bool(key, value),

123 > }

124 > }

125

126 func NewErrorTag(key string, value error) ZapTag {

154 }

155

156 > func NewAnyTag(key string, value any) ZapTag { zap_tag.go

157 > return ZapTag{

158 > field: zap.Any(key, value),

159 > }

160 > }

161

162 func NewBinaryTag(key string, value []byte) ZapTag {

224 }

225

226 > func Any(key string, value any) ZapTag { zap_tag.go

227 > return NewAnyTag(key, value)

228 > }

229

230 func Binary(key string, value []byte) ZapTag {

go.temporal.io/server/api/enums/v1/predicate.pb.go 23 covered LOC · 3 ranges

Open complete file

111 }

112

113 > func (PredicateType) Descriptor() protoreflect.EnumDescriptor { predicate.pb.go

114 > return file_temporal_server_api_enums_v1_predicate_proto_enumTypes[0].Descriptor()

115 > }

116

117 func (PredicateType) Type() protoreflect.EnumType {

170 }

171

172 > func init() { file_temporal_server_api_enums_v1_predicate_proto_init() } predicate.pb.go

173 > func file_temporal_server_api_enums_v1_predicate_proto_init() {

174 > if File_temporal_server_api_enums_v1_predicate_proto != nil {

175 return

176 }

177 > type x struct{} predicate.pb.go

178 > out := protoimpl.TypeBuilder{

179 > File: protoimpl.DescBuilder{

180 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

181 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_predicate_proto_rawDesc), len(file_temporal_server_api_enums_v1_predicate_proto_rawDesc)),

182 > NumEnums: 1,

183 > NumMessages: 0,

184 > NumExtensions: 0,

185 > NumServices: 0,

186 > },

187 > GoTypes: file_temporal_server_api_enums_v1_predicate_proto_goTypes,

188 > DependencyIndexes: file_temporal_server_api_enums_v1_predicate_proto_depIdxs,

189 > EnumInfos: file_temporal_server_api_enums_v1_predicate_proto_enumTypes,

190 > }.Build()

191 > File_temporal_server_api_enums_v1_predicate_proto = out.File

192 > file_temporal_server_api_enums_v1_predicate_proto_goTypes = nil

193 > file_temporal_server_api_enums_v1_predicate_proto_depIdxs = nil

194 }

go.temporal.io/server/api/enums/v1/task.pb.go 23 covered LOC · 3 ranges

Open complete file

303 }

304

305 > func (TaskType) Descriptor() protoreflect.EnumDescriptor { task.pb.go

306 > return file_temporal_server_api_enums_v1_task_proto_enumTypes[1].Descriptor()

307 > }

308

309 func (TaskType) Type() protoreflect.EnumType {

456 }

457

458 > func init() { file_temporal_server_api_enums_v1_task_proto_init() } task.pb.go

459 > func file_temporal_server_api_enums_v1_task_proto_init() {

460 > if File_temporal_server_api_enums_v1_task_proto != nil {

461 return

462 }

463 > type x struct{} task.pb.go

464 > out := protoimpl.TypeBuilder{

465 > File: protoimpl.DescBuilder{

466 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

467 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_task_proto_rawDesc), len(file_temporal_server_api_enums_v1_task_proto_rawDesc)),

468 > NumEnums: 3,

469 > NumMessages: 0,

470 > NumExtensions: 0,

471 > NumServices: 0,

472 > },

473 > GoTypes: file_temporal_server_api_enums_v1_task_proto_goTypes,

474 > DependencyIndexes: file_temporal_server_api_enums_v1_task_proto_depIdxs,

475 > EnumInfos: file_temporal_server_api_enums_v1_task_proto_enumTypes,

476 > }.Build()

477 > File_temporal_server_api_enums_v1_task_proto = out.File

478 > file_temporal_server_api_enums_v1_task_proto_goTypes = nil

479 > file_temporal_server_api_enums_v1_task_proto_depIdxs = nil

480 }

go.temporal.io/server/api/persistence/v1/queues.pb.go 23 covered LOC · 1 range

Open complete file

616 }

617

618 > func init() { file_temporal_server_api_persistence_v1_queues_proto_init() } queues.pb.go

619 > func file_temporal_server_api_persistence_v1_queues_proto_init() {

620 > if File_temporal_server_api_persistence_v1_queues_proto != nil {

621 > return

622 > }

623 > file_temporal_server_api_persistence_v1_predicates_proto_init()

624 > type x struct{}

625 > out := protoimpl.TypeBuilder{

626 > File: protoimpl.DescBuilder{

627 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

628 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_queues_proto_rawDesc), len(file_temporal_server_api_persistence_v1_queues_proto_rawDesc)),

629 > NumEnums: 0,

630 > NumMessages: 12,

631 > NumExtensions: 0,

632 > NumServices: 0,

633 > },

634 > GoTypes: file_temporal_server_api_persistence_v1_queues_proto_goTypes,

635 > DependencyIndexes: file_temporal_server_api_persistence_v1_queues_proto_depIdxs,

636 > MessageInfos: file_temporal_server_api_persistence_v1_queues_proto_msgTypes,

637 > }.Build()

638 > File_temporal_server_api_persistence_v1_queues_proto = out.File

639 > file_temporal_server_api_persistence_v1_queues_proto_goTypes = nil

640 > file_temporal_server_api_persistence_v1_queues_proto_depIdxs = nil

641 }

go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1/request_response.pb.go 23 covered LOC · 1 range

Open complete file

1021 }

1022

1023 > func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_init() } request_response.pb.go

1024 > func file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_init() {

1025 > if File_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto != nil {

1026 > return

1027 > }

1028 > file_temporal_server_chasm_lib_scheduler_proto_v1_message_proto_init()

1029 > type x struct{}

1030 > out := protoimpl.TypeBuilder{

1031 > File: protoimpl.DescBuilder{

1032 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

1033 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_rawDesc)),

1034 > NumEnums: 0,

1035 > NumMessages: 18,

1036 > NumExtensions: 0,

1037 > NumServices: 0,

1038 > },

1039 > GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_goTypes,

1040 > DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_depIdxs,

1041 > MessageInfos: file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_msgTypes,

1042 > }.Build()

1043 > File_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto = out.File

1044 > file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_goTypes = nil

1045 > file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_depIdxs = nil

1046 }

go.temporal.io/server/api/common/v1/api_category.pb.go 22 covered LOC · 2 ranges

Open complete file

204 }

205

206 > func init() { file_temporal_server_api_common_v1_api_category_proto_init() } api_category.pb.go

207 > func file_temporal_server_api_common_v1_api_category_proto_init() {

208 > if File_temporal_server_api_common_v1_api_category_proto != nil {

209 return

210 }

211 > type x struct{} api_category.pb.go

212 > out := protoimpl.TypeBuilder{

213 > File: protoimpl.DescBuilder{

214 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

215 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_common_v1_api_category_proto_rawDesc), len(file_temporal_server_api_common_v1_api_category_proto_rawDesc)),

216 > NumEnums: 1,

217 > NumMessages: 1,

218 > NumExtensions: 1,

219 > NumServices: 0,

220 > },

221 > GoTypes: file_temporal_server_api_common_v1_api_category_proto_goTypes,

222 > DependencyIndexes: file_temporal_server_api_common_v1_api_category_proto_depIdxs,

223 > EnumInfos: file_temporal_server_api_common_v1_api_category_proto_enumTypes,

224 > MessageInfos: file_temporal_server_api_common_v1_api_category_proto_msgTypes,

225 > ExtensionInfos: file_temporal_server_api_common_v1_api_category_proto_extTypes,

226 > }.Build()

227 > File_temporal_server_api_common_v1_api_category_proto = out.File

228 > file_temporal_server_api_common_v1_api_category_proto_goTypes = nil

229 > file_temporal_server_api_common_v1_api_category_proto_depIdxs = nil

230 }

go.temporal.io/server/api/persistence/v1/task_queues.pb.go 21 covered LOC · 2 ranges

Open complete file

899 }

900

901 > func init() { file_temporal_server_api_persistence_v1_task_queues_proto_init() } task_queues.pb.go

902 > func file_temporal_server_api_persistence_v1_task_queues_proto_init() {

903 > if File_temporal_server_api_persistence_v1_task_queues_proto != nil {

904 return

905 }

906 > type x struct{} task_queues.pb.go

907 > out := protoimpl.TypeBuilder{

908 > File: protoimpl.DescBuilder{

909 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

910 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_task_queues_proto_rawDesc), len(file_temporal_server_api_persistence_v1_task_queues_proto_rawDesc)),

911 > NumEnums: 1,

912 > NumMessages: 13,

913 > NumExtensions: 0,

914 > NumServices: 0,

915 > },

916 > GoTypes: file_temporal_server_api_persistence_v1_task_queues_proto_goTypes,

917 > DependencyIndexes: file_temporal_server_api_persistence_v1_task_queues_proto_depIdxs,

918 > EnumInfos: file_temporal_server_api_persistence_v1_task_queues_proto_enumTypes,

919 > MessageInfos: file_temporal_server_api_persistence_v1_task_queues_proto_msgTypes,

920 > }.Build()

921 > File_temporal_server_api_persistence_v1_task_queues_proto = out.File

922 > file_temporal_server_api_persistence_v1_task_queues_proto_goTypes = nil

923 > file_temporal_server_api_persistence_v1_task_queues_proto_depIdxs = nil

924 }

go.temporal.io/server/api/routing/v1/extension.pb.go 21 covered LOC · 2 ranges

Open complete file

144 }

145

146 > func init() { file_temporal_server_api_routing_v1_extension_proto_init() } extension.pb.go

147 > func file_temporal_server_api_routing_v1_extension_proto_init() {

148 > if File_temporal_server_api_routing_v1_extension_proto != nil {

149 return

150 }

151 > type x struct{} extension.pb.go

152 > out := protoimpl.TypeBuilder{

153 > File: protoimpl.DescBuilder{

154 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

155 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_routing_v1_extension_proto_rawDesc), len(file_temporal_server_api_routing_v1_extension_proto_rawDesc)),

156 > NumEnums: 0,

157 > NumMessages: 1,

158 > NumExtensions: 1,

159 > NumServices: 0,

160 > },

161 > GoTypes: file_temporal_server_api_routing_v1_extension_proto_goTypes,

162 > DependencyIndexes: file_temporal_server_api_routing_v1_extension_proto_depIdxs,

163 > MessageInfos: file_temporal_server_api_routing_v1_extension_proto_msgTypes,

164 > ExtensionInfos: file_temporal_server_api_routing_v1_extension_proto_extTypes,

165 > }.Build()

166 > File_temporal_server_api_routing_v1_extension_proto = out.File

167 > file_temporal_server_api_routing_v1_extension_proto_goTypes = nil

168 > file_temporal_server_api_routing_v1_extension_proto_depIdxs = nil

169 }

go.temporal.io/server/api/adminservice/v1/service.pb.go 20 covered LOC · 2 ranges

Open complete file

273 }

274

275 > func init() { file_temporal_server_api_adminservice_v1_service_proto_init() } service.pb.go

276 > func file_temporal_server_api_adminservice_v1_service_proto_init() {

277 > if File_temporal_server_api_adminservice_v1_service_proto != nil {

278 return

279 }

280 > file_temporal_server_api_adminservice_v1_request_response_proto_init() service.pb.go

281 > type x struct{}

282 > out := protoimpl.TypeBuilder{

283 > File: protoimpl.DescBuilder{

284 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

285 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_adminservice_v1_service_proto_rawDesc), len(file_temporal_server_api_adminservice_v1_service_proto_rawDesc)),

286 > NumEnums: 0,

287 > NumMessages: 0,

288 > NumExtensions: 0,

289 > NumServices: 1,

290 > },

291 > GoTypes: file_temporal_server_api_adminservice_v1_service_proto_goTypes,

292 > DependencyIndexes: file_temporal_server_api_adminservice_v1_service_proto_depIdxs,

293 > }.Build()

294 > File_temporal_server_api_adminservice_v1_service_proto = out.File

295 > file_temporal_server_api_adminservice_v1_service_proto_goTypes = nil

296 > file_temporal_server_api_adminservice_v1_service_proto_depIdxs = nil

297 }

go.temporal.io/server/api/archiver/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

431 }

432

433 > func init() { file_temporal_server_api_archiver_v1_message_proto_init() } message.pb.go

434 > func file_temporal_server_api_archiver_v1_message_proto_init() {

435 > if File_temporal_server_api_archiver_v1_message_proto != nil {

436 return

437 }

438 > type x struct{} message.pb.go

439 > out := protoimpl.TypeBuilder{

440 > File: protoimpl.DescBuilder{

441 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

442 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_archiver_v1_message_proto_rawDesc), len(file_temporal_server_api_archiver_v1_message_proto_rawDesc)),

443 > NumEnums: 0,

444 > NumMessages: 4,

445 > NumExtensions: 0,

446 > NumServices: 0,

447 > },

448 > GoTypes: file_temporal_server_api_archiver_v1_message_proto_goTypes,

449 > DependencyIndexes: file_temporal_server_api_archiver_v1_message_proto_depIdxs,

450 > MessageInfos: file_temporal_server_api_archiver_v1_message_proto_msgTypes,

451 > }.Build()

452 > File_temporal_server_api_archiver_v1_message_proto = out.File

453 > file_temporal_server_api_archiver_v1_message_proto_goTypes = nil

454 > file_temporal_server_api_archiver_v1_message_proto_depIdxs = nil

455 }

go.temporal.io/server/api/chasm/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

206 }

207

208 > func init() { file_temporal_server_api_chasm_v1_message_proto_init() } message.pb.go

209 > func file_temporal_server_api_chasm_v1_message_proto_init() {

210 > if File_temporal_server_api_chasm_v1_message_proto != nil {

211 return

212 }

213 > type x struct{} message.pb.go

214 > out := protoimpl.TypeBuilder{

215 > File: protoimpl.DescBuilder{

216 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

217 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_chasm_v1_message_proto_rawDesc), len(file_temporal_server_api_chasm_v1_message_proto_rawDesc)),

218 > NumEnums: 0,

219 > NumMessages: 1,

220 > NumExtensions: 0,

221 > NumServices: 0,

222 > },

223 > GoTypes: file_temporal_server_api_chasm_v1_message_proto_goTypes,

224 > DependencyIndexes: file_temporal_server_api_chasm_v1_message_proto_depIdxs,

225 > MessageInfos: file_temporal_server_api_chasm_v1_message_proto_msgTypes,

226 > }.Build()

227 > File_temporal_server_api_chasm_v1_message_proto = out.File

228 > file_temporal_server_api_chasm_v1_message_proto_goTypes = nil

229 > file_temporal_server_api_chasm_v1_message_proto_depIdxs = nil

230 }

go.temporal.io/server/api/clock/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

193 }

194

195 > func init() { file_temporal_server_api_clock_v1_message_proto_init() } message.pb.go

196 > func file_temporal_server_api_clock_v1_message_proto_init() {

197 > if File_temporal_server_api_clock_v1_message_proto != nil {

198 return

199 }

200 > type x struct{} message.pb.go

201 > out := protoimpl.TypeBuilder{

202 > File: protoimpl.DescBuilder{

203 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

204 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_clock_v1_message_proto_rawDesc), len(file_temporal_server_api_clock_v1_message_proto_rawDesc)),

205 > NumEnums: 0,

206 > NumMessages: 2,

207 > NumExtensions: 0,

208 > NumServices: 0,

209 > },

210 > GoTypes: file_temporal_server_api_clock_v1_message_proto_goTypes,

211 > DependencyIndexes: file_temporal_server_api_clock_v1_message_proto_depIdxs,

212 > MessageInfos: file_temporal_server_api_clock_v1_message_proto_msgTypes,

213 > }.Build()

214 > File_temporal_server_api_clock_v1_message_proto = out.File

215 > file_temporal_server_api_clock_v1_message_proto_goTypes = nil

216 > file_temporal_server_api_clock_v1_message_proto_depIdxs = nil

217 }

go.temporal.io/server/api/cluster/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

342 }

343

344 > func init() { file_temporal_server_api_cluster_v1_message_proto_init() } message.pb.go

345 > func file_temporal_server_api_cluster_v1_message_proto_init() {

346 > if File_temporal_server_api_cluster_v1_message_proto != nil {

347 return

348 }

349 > type x struct{} message.pb.go

350 > out := protoimpl.TypeBuilder{

351 > File: protoimpl.DescBuilder{

352 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

353 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_cluster_v1_message_proto_rawDesc), len(file_temporal_server_api_cluster_v1_message_proto_rawDesc)),

354 > NumEnums: 0,

355 > NumMessages: 4,

356 > NumExtensions: 0,

357 > NumServices: 0,

358 > },

359 > GoTypes: file_temporal_server_api_cluster_v1_message_proto_goTypes,

360 > DependencyIndexes: file_temporal_server_api_cluster_v1_message_proto_depIdxs,

361 > MessageInfos: file_temporal_server_api_cluster_v1_message_proto_msgTypes,

362 > }.Build()

363 > File_temporal_server_api_cluster_v1_message_proto = out.File

364 > file_temporal_server_api_cluster_v1_message_proto_goTypes = nil

365 > file_temporal_server_api_cluster_v1_message_proto_depIdxs = nil

366 }

go.temporal.io/server/api/common/v1/dlq.pb.go 20 covered LOC · 2 ranges

Open complete file

295 }

296

297 > func init() { file_temporal_server_api_common_v1_dlq_proto_init() } dlq.pb.go

298 > func file_temporal_server_api_common_v1_dlq_proto_init() {

299 > if File_temporal_server_api_common_v1_dlq_proto != nil {

300 return

301 }

302 > type x struct{} dlq.pb.go

303 > out := protoimpl.TypeBuilder{

304 > File: protoimpl.DescBuilder{

305 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

306 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_common_v1_dlq_proto_rawDesc), len(file_temporal_server_api_common_v1_dlq_proto_rawDesc)),

307 > NumEnums: 0,

308 > NumMessages: 4,

309 > NumExtensions: 0,

310 > NumServices: 0,

311 > },

312 > GoTypes: file_temporal_server_api_common_v1_dlq_proto_goTypes,

313 > DependencyIndexes: file_temporal_server_api_common_v1_dlq_proto_depIdxs,

314 > MessageInfos: file_temporal_server_api_common_v1_dlq_proto_msgTypes,

315 > }.Build()

316 > File_temporal_server_api_common_v1_dlq_proto = out.File

317 > file_temporal_server_api_common_v1_dlq_proto_goTypes = nil

318 > file_temporal_server_api_common_v1_dlq_proto_depIdxs = nil

319 }

go.temporal.io/server/api/contextpropagation/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

109 }

110

111 > func init() { file_temporal_server_api_contextpropagation_v1_message_proto_init() } message.pb.go

112 > func file_temporal_server_api_contextpropagation_v1_message_proto_init() {

113 > if File_temporal_server_api_contextpropagation_v1_message_proto != nil {

114 return

115 }

116 > type x struct{} message.pb.go

117 > out := protoimpl.TypeBuilder{

118 > File: protoimpl.DescBuilder{

119 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

120 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_contextpropagation_v1_message_proto_rawDesc), len(file_temporal_server_api_contextpropagation_v1_message_proto_rawDesc)),

121 > NumEnums: 0,

122 > NumMessages: 2,

123 > NumExtensions: 0,

124 > NumServices: 0,

125 > },

126 > GoTypes: file_temporal_server_api_contextpropagation_v1_message_proto_goTypes,

127 > DependencyIndexes: file_temporal_server_api_contextpropagation_v1_message_proto_depIdxs,

128 > MessageInfos: file_temporal_server_api_contextpropagation_v1_message_proto_msgTypes,

129 > }.Build()

130 > File_temporal_server_api_contextpropagation_v1_message_proto = out.File

131 > file_temporal_server_api_contextpropagation_v1_message_proto_goTypes = nil

132 > file_temporal_server_api_contextpropagation_v1_message_proto_depIdxs = nil

133 }

go.temporal.io/server/api/deployment/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

4626 }

4627

4628 > func init() { file_temporal_server_api_deployment_v1_message_proto_init() } message.pb.go

4629 > func file_temporal_server_api_deployment_v1_message_proto_init() {

4630 > if File_temporal_server_api_deployment_v1_message_proto != nil {

4631 return

4632 }

4633 > type x struct{} message.pb.go

4634 > out := protoimpl.TypeBuilder{

4635 > File: protoimpl.DescBuilder{

4636 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

4637 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_deployment_v1_message_proto_rawDesc), len(file_temporal_server_api_deployment_v1_message_proto_rawDesc)),

4638 > NumEnums: 0,

4639 > NumMessages: 75,

4640 > NumExtensions: 0,

4641 > NumServices: 0,

4642 > },

4643 > GoTypes: file_temporal_server_api_deployment_v1_message_proto_goTypes,

4644 > DependencyIndexes: file_temporal_server_api_deployment_v1_message_proto_depIdxs,

4645 > MessageInfos: file_temporal_server_api_deployment_v1_message_proto_msgTypes,

4646 > }.Build()

4647 > File_temporal_server_api_deployment_v1_message_proto = out.File

4648 > file_temporal_server_api_deployment_v1_message_proto_goTypes = nil

4649 > file_temporal_server_api_deployment_v1_message_proto_depIdxs = nil

4650 }

go.temporal.io/server/api/enums/v1/cluster.pb.go 20 covered LOC · 2 ranges

Open complete file

209 }

210

211 > func init() { file_temporal_server_api_enums_v1_cluster_proto_init() } cluster.pb.go

212 > func file_temporal_server_api_enums_v1_cluster_proto_init() {

213 > if File_temporal_server_api_enums_v1_cluster_proto != nil {

214 return

215 }

216 > type x struct{} cluster.pb.go

217 > out := protoimpl.TypeBuilder{

218 > File: protoimpl.DescBuilder{

219 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

220 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_cluster_proto_rawDesc), len(file_temporal_server_api_enums_v1_cluster_proto_rawDesc)),

221 > NumEnums: 2,

222 > NumMessages: 0,

223 > NumExtensions: 0,

224 > NumServices: 0,

225 > },

226 > GoTypes: file_temporal_server_api_enums_v1_cluster_proto_goTypes,

227 > DependencyIndexes: file_temporal_server_api_enums_v1_cluster_proto_depIdxs,

228 > EnumInfos: file_temporal_server_api_enums_v1_cluster_proto_enumTypes,

229 > }.Build()

230 > File_temporal_server_api_enums_v1_cluster_proto = out.File

231 > file_temporal_server_api_enums_v1_cluster_proto_goTypes = nil

232 > file_temporal_server_api_enums_v1_cluster_proto_depIdxs = nil

233 }

go.temporal.io/server/api/enums/v1/common.pb.go 20 covered LOC · 2 ranges

Open complete file

264 }

265

266 > func init() { file_temporal_server_api_enums_v1_common_proto_init() } common.pb.go

267 > func file_temporal_server_api_enums_v1_common_proto_init() {

268 > if File_temporal_server_api_enums_v1_common_proto != nil {

269 return

270 }

271 > type x struct{} common.pb.go

272 > out := protoimpl.TypeBuilder{

273 > File: protoimpl.DescBuilder{

274 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

275 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_common_proto_rawDesc), len(file_temporal_server_api_enums_v1_common_proto_rawDesc)),

276 > NumEnums: 3,

277 > NumMessages: 0,

278 > NumExtensions: 0,

279 > NumServices: 0,

280 > },

281 > GoTypes: file_temporal_server_api_enums_v1_common_proto_goTypes,

282 > DependencyIndexes: file_temporal_server_api_enums_v1_common_proto_depIdxs,

283 > EnumInfos: file_temporal_server_api_enums_v1_common_proto_enumTypes,

284 > }.Build()

285 > File_temporal_server_api_enums_v1_common_proto = out.File

286 > file_temporal_server_api_enums_v1_common_proto_goTypes = nil

287 > file_temporal_server_api_enums_v1_common_proto_depIdxs = nil

288 }

go.temporal.io/server/api/enums/v1/dlq.pb.go 20 covered LOC · 2 ranges

Open complete file

187 }

188

189 > func init() { file_temporal_server_api_enums_v1_dlq_proto_init() } dlq.pb.go

190 > func file_temporal_server_api_enums_v1_dlq_proto_init() {

191 > if File_temporal_server_api_enums_v1_dlq_proto != nil {

192 return

193 }

194 > type x struct{} dlq.pb.go

195 > out := protoimpl.TypeBuilder{

196 > File: protoimpl.DescBuilder{

197 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

198 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_dlq_proto_rawDesc), len(file_temporal_server_api_enums_v1_dlq_proto_rawDesc)),

199 > NumEnums: 2,

200 > NumMessages: 0,

201 > NumExtensions: 0,

202 > NumServices: 0,

203 > },

204 > GoTypes: file_temporal_server_api_enums_v1_dlq_proto_goTypes,

205 > DependencyIndexes: file_temporal_server_api_enums_v1_dlq_proto_depIdxs,

206 > EnumInfos: file_temporal_server_api_enums_v1_dlq_proto_enumTypes,

207 > }.Build()

208 > File_temporal_server_api_enums_v1_dlq_proto = out.File

209 > file_temporal_server_api_enums_v1_dlq_proto_goTypes = nil

210 > file_temporal_server_api_enums_v1_dlq_proto_depIdxs = nil

211 }

go.temporal.io/server/api/enums/v1/fairness_state.pb.go 20 covered LOC · 2 ranges

Open complete file

123 }

124

125 > func init() { file_temporal_server_api_enums_v1_fairness_state_proto_init() } fairness_state.pb.go

126 > func file_temporal_server_api_enums_v1_fairness_state_proto_init() {

127 > if File_temporal_server_api_enums_v1_fairness_state_proto != nil {

128 return

129 }

130 > type x struct{} fairness_state.pb.go

131 > out := protoimpl.TypeBuilder{

132 > File: protoimpl.DescBuilder{

133 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

134 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_fairness_state_proto_rawDesc), len(file_temporal_server_api_enums_v1_fairness_state_proto_rawDesc)),

135 > NumEnums: 1,

136 > NumMessages: 0,

137 > NumExtensions: 0,

138 > NumServices: 0,

139 > },

140 > GoTypes: file_temporal_server_api_enums_v1_fairness_state_proto_goTypes,

141 > DependencyIndexes: file_temporal_server_api_enums_v1_fairness_state_proto_depIdxs,

142 > EnumInfos: file_temporal_server_api_enums_v1_fairness_state_proto_enumTypes,

143 > }.Build()

144 > File_temporal_server_api_enums_v1_fairness_state_proto = out.File

145 > file_temporal_server_api_enums_v1_fairness_state_proto_goTypes = nil

146 > file_temporal_server_api_enums_v1_fairness_state_proto_depIdxs = nil

147 }

go.temporal.io/server/api/enums/v1/nexus.pb.go 20 covered LOC · 2 ranges

Open complete file

158 }

159

160 > func init() { file_temporal_server_api_enums_v1_nexus_proto_init() } nexus.pb.go

161 > func file_temporal_server_api_enums_v1_nexus_proto_init() {

162 > if File_temporal_server_api_enums_v1_nexus_proto != nil {

163 return

164 }

165 > type x struct{} nexus.pb.go

166 > out := protoimpl.TypeBuilder{

167 > File: protoimpl.DescBuilder{

168 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

169 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_nexus_proto_rawDesc), len(file_temporal_server_api_enums_v1_nexus_proto_rawDesc)),

170 > NumEnums: 1,

171 > NumMessages: 0,

172 > NumExtensions: 0,

173 > NumServices: 0,

174 > },

175 > GoTypes: file_temporal_server_api_enums_v1_nexus_proto_goTypes,

176 > DependencyIndexes: file_temporal_server_api_enums_v1_nexus_proto_depIdxs,

177 > EnumInfos: file_temporal_server_api_enums_v1_nexus_proto_enumTypes,

178 > }.Build()

179 > File_temporal_server_api_enums_v1_nexus_proto = out.File

180 > file_temporal_server_api_enums_v1_nexus_proto_goTypes = nil

181 > file_temporal_server_api_enums_v1_nexus_proto_depIdxs = nil

182 }

go.temporal.io/server/api/enums/v1/replication.pb.go 20 covered LOC · 2 ranges

Open complete file

314 }

315

316 > func init() { file_temporal_server_api_enums_v1_replication_proto_init() } replication.pb.go

317 > func file_temporal_server_api_enums_v1_replication_proto_init() {

318 > if File_temporal_server_api_enums_v1_replication_proto != nil {

319 return

320 }

321 > type x struct{} replication.pb.go

322 > out := protoimpl.TypeBuilder{

323 > File: protoimpl.DescBuilder{

324 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

325 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_replication_proto_rawDesc), len(file_temporal_server_api_enums_v1_replication_proto_rawDesc)),

326 > NumEnums: 3,

327 > NumMessages: 0,

328 > NumExtensions: 0,

329 > NumServices: 0,

330 > },

331 > GoTypes: file_temporal_server_api_enums_v1_replication_proto_goTypes,

332 > DependencyIndexes: file_temporal_server_api_enums_v1_replication_proto_depIdxs,

333 > EnumInfos: file_temporal_server_api_enums_v1_replication_proto_enumTypes,

334 > }.Build()

335 > File_temporal_server_api_enums_v1_replication_proto = out.File

336 > file_temporal_server_api_enums_v1_replication_proto_goTypes = nil

337 > file_temporal_server_api_enums_v1_replication_proto_depIdxs = nil

338 }

go.temporal.io/server/api/enums/v1/workflow.pb.go 20 covered LOC · 2 ranges

Open complete file

275 }

276

277 > func init() { file_temporal_server_api_enums_v1_workflow_proto_init() } workflow.pb.go

278 > func file_temporal_server_api_enums_v1_workflow_proto_init() {

279 > if File_temporal_server_api_enums_v1_workflow_proto != nil {

280 return

281 }

282 > type x struct{} workflow.pb.go

283 > out := protoimpl.TypeBuilder{

284 > File: protoimpl.DescBuilder{

285 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

286 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_workflow_proto_rawDesc), len(file_temporal_server_api_enums_v1_workflow_proto_rawDesc)),

287 > NumEnums: 3,

288 > NumMessages: 0,

289 > NumExtensions: 0,

290 > NumServices: 0,

291 > },

292 > GoTypes: file_temporal_server_api_enums_v1_workflow_proto_goTypes,

293 > DependencyIndexes: file_temporal_server_api_enums_v1_workflow_proto_depIdxs,

294 > EnumInfos: file_temporal_server_api_enums_v1_workflow_proto_enumTypes,

295 > }.Build()

296 > File_temporal_server_api_enums_v1_workflow_proto = out.File

297 > file_temporal_server_api_enums_v1_workflow_proto_goTypes = nil

298 > file_temporal_server_api_enums_v1_workflow_proto_depIdxs = nil

299 }

go.temporal.io/server/api/enums/v1/workflow_task_type.pb.go 20 covered LOC · 2 ranges

Open complete file

124 }

125

126 > func init() { file_temporal_server_api_enums_v1_workflow_task_type_proto_init() } workflow_task_type.pb.go

127 > func file_temporal_server_api_enums_v1_workflow_task_type_proto_init() {

128 > if File_temporal_server_api_enums_v1_workflow_task_type_proto != nil {

129 return

130 }

131 > type x struct{} workflow_task_type.pb.go

132 > out := protoimpl.TypeBuilder{

133 > File: protoimpl.DescBuilder{

134 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

135 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_enums_v1_workflow_task_type_proto_rawDesc), len(file_temporal_server_api_enums_v1_workflow_task_type_proto_rawDesc)),

136 > NumEnums: 1,

137 > NumMessages: 0,

138 > NumExtensions: 0,

139 > NumServices: 0,

140 > },

141 > GoTypes: file_temporal_server_api_enums_v1_workflow_task_type_proto_goTypes,

142 > DependencyIndexes: file_temporal_server_api_enums_v1_workflow_task_type_proto_depIdxs,

143 > EnumInfos: file_temporal_server_api_enums_v1_workflow_task_type_proto_enumTypes,

144 > }.Build()

145 > File_temporal_server_api_enums_v1_workflow_task_type_proto = out.File

146 > file_temporal_server_api_enums_v1_workflow_task_type_proto_goTypes = nil

147 > file_temporal_server_api_enums_v1_workflow_task_type_proto_depIdxs = nil

148 }

go.temporal.io/server/api/errordetails/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

625 }

626

627 > func init() { file_temporal_server_api_errordetails_v1_message_proto_init() } message.pb.go

628 > func file_temporal_server_api_errordetails_v1_message_proto_init() {

629 > if File_temporal_server_api_errordetails_v1_message_proto != nil {

630 return

631 }

632 > type x struct{} message.pb.go

633 > out := protoimpl.TypeBuilder{

634 > File: protoimpl.DescBuilder{

635 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

636 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_errordetails_v1_message_proto_rawDesc), len(file_temporal_server_api_errordetails_v1_message_proto_rawDesc)),

637 > NumEnums: 0,

638 > NumMessages: 10,

639 > NumExtensions: 0,

640 > NumServices: 0,

641 > },

642 > GoTypes: file_temporal_server_api_errordetails_v1_message_proto_goTypes,

643 > DependencyIndexes: file_temporal_server_api_errordetails_v1_message_proto_depIdxs,

644 > MessageInfos: file_temporal_server_api_errordetails_v1_message_proto_msgTypes,

645 > }.Build()

646 > File_temporal_server_api_errordetails_v1_message_proto = out.File

647 > file_temporal_server_api_errordetails_v1_message_proto_goTypes = nil

648 > file_temporal_server_api_errordetails_v1_message_proto_depIdxs = nil

649 }

go.temporal.io/server/api/health/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

300 }

301

302 > func init() { file_temporal_server_api_health_v1_message_proto_init() } message.pb.go

303 > func file_temporal_server_api_health_v1_message_proto_init() {

304 > if File_temporal_server_api_health_v1_message_proto != nil {

305 return

306 }

307 > type x struct{} message.pb.go

308 > out := protoimpl.TypeBuilder{

309 > File: protoimpl.DescBuilder{

310 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

311 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_health_v1_message_proto_rawDesc), len(file_temporal_server_api_health_v1_message_proto_rawDesc)),

312 > NumEnums: 0,

313 > NumMessages: 3,

314 > NumExtensions: 0,

315 > NumServices: 0,

316 > },

317 > GoTypes: file_temporal_server_api_health_v1_message_proto_goTypes,

318 > DependencyIndexes: file_temporal_server_api_health_v1_message_proto_depIdxs,

319 > MessageInfos: file_temporal_server_api_health_v1_message_proto_msgTypes,

320 > }.Build()

321 > File_temporal_server_api_health_v1_message_proto = out.File

322 > file_temporal_server_api_health_v1_message_proto_goTypes = nil

323 > file_temporal_server_api_health_v1_message_proto_depIdxs = nil

324 }

go.temporal.io/server/api/history/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

498 }

499

500 > func init() { file_temporal_server_api_history_v1_message_proto_init() } message.pb.go

501 > func file_temporal_server_api_history_v1_message_proto_init() {

502 > if File_temporal_server_api_history_v1_message_proto != nil {

503 return

504 }

505 > type x struct{} message.pb.go

506 > out := protoimpl.TypeBuilder{

507 > File: protoimpl.DescBuilder{

508 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

509 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_history_v1_message_proto_rawDesc), len(file_temporal_server_api_history_v1_message_proto_rawDesc)),

510 > NumEnums: 0,

511 > NumMessages: 8,

512 > NumExtensions: 0,

513 > NumServices: 0,

514 > },

515 > GoTypes: file_temporal_server_api_history_v1_message_proto_goTypes,

516 > DependencyIndexes: file_temporal_server_api_history_v1_message_proto_depIdxs,

517 > MessageInfos: file_temporal_server_api_history_v1_message_proto_msgTypes,

518 > }.Build()

519 > File_temporal_server_api_history_v1_message_proto = out.File

520 > file_temporal_server_api_history_v1_message_proto_goTypes = nil

521 > file_temporal_server_api_history_v1_message_proto_depIdxs = nil

522 }

go.temporal.io/server/api/historyservice/v1/service.pb.go 20 covered LOC · 2 ranges

Open complete file

428 }

429

430 > func init() { file_temporal_server_api_historyservice_v1_service_proto_init() } service.pb.go

431 > func file_temporal_server_api_historyservice_v1_service_proto_init() {

432 > if File_temporal_server_api_historyservice_v1_service_proto != nil {

433 return

434 }

435 > file_temporal_server_api_historyservice_v1_request_response_proto_init() service.pb.go

436 > type x struct{}

437 > out := protoimpl.TypeBuilder{

438 > File: protoimpl.DescBuilder{

439 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

440 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_historyservice_v1_service_proto_rawDesc), len(file_temporal_server_api_historyservice_v1_service_proto_rawDesc)),

441 > NumEnums: 0,

442 > NumMessages: 0,

443 > NumExtensions: 0,

444 > NumServices: 1,

445 > },

446 > GoTypes: file_temporal_server_api_historyservice_v1_service_proto_goTypes,

447 > DependencyIndexes: file_temporal_server_api_historyservice_v1_service_proto_depIdxs,

448 > }.Build()

449 > File_temporal_server_api_historyservice_v1_service_proto = out.File

450 > file_temporal_server_api_historyservice_v1_service_proto_goTypes = nil

451 > file_temporal_server_api_historyservice_v1_service_proto_depIdxs = nil

452 }

go.temporal.io/server/api/matchingservice/v1/service.pb.go 20 covered LOC · 2 ranges

Open complete file

250 }

251

252 > func init() { file_temporal_server_api_matchingservice_v1_service_proto_init() } service.pb.go

253 > func file_temporal_server_api_matchingservice_v1_service_proto_init() {

254 > if File_temporal_server_api_matchingservice_v1_service_proto != nil {

255 return

256 }

257 > file_temporal_server_api_matchingservice_v1_request_response_proto_init() service.pb.go

258 > type x struct{}

259 > out := protoimpl.TypeBuilder{

260 > File: protoimpl.DescBuilder{

261 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

262 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_matchingservice_v1_service_proto_rawDesc), len(file_temporal_server_api_matchingservice_v1_service_proto_rawDesc)),

263 > NumEnums: 0,

264 > NumMessages: 0,

265 > NumExtensions: 0,

266 > NumServices: 1,

267 > },

268 > GoTypes: file_temporal_server_api_matchingservice_v1_service_proto_goTypes,

269 > DependencyIndexes: file_temporal_server_api_matchingservice_v1_service_proto_depIdxs,

270 > }.Build()

271 > File_temporal_server_api_matchingservice_v1_service_proto = out.File

272 > file_temporal_server_api_matchingservice_v1_service_proto_goTypes = nil

273 > file_temporal_server_api_matchingservice_v1_service_proto_depIdxs = nil

274 }

go.temporal.io/server/api/metrics/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

104 }

105

106 > func init() { file_temporal_server_api_metrics_v1_message_proto_init() } message.pb.go

107 > func file_temporal_server_api_metrics_v1_message_proto_init() {

108 > if File_temporal_server_api_metrics_v1_message_proto != nil {

109 return

110 }

111 > type x struct{} message.pb.go

112 > out := protoimpl.TypeBuilder{

113 > File: protoimpl.DescBuilder{

114 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

115 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_metrics_v1_message_proto_rawDesc), len(file_temporal_server_api_metrics_v1_message_proto_rawDesc)),

116 > NumEnums: 0,

117 > NumMessages: 2,

118 > NumExtensions: 0,

119 > NumServices: 0,

120 > },

121 > GoTypes: file_temporal_server_api_metrics_v1_message_proto_goTypes,

122 > DependencyIndexes: file_temporal_server_api_metrics_v1_message_proto_depIdxs,

123 > MessageInfos: file_temporal_server_api_metrics_v1_message_proto_msgTypes,

124 > }.Build()

125 > File_temporal_server_api_metrics_v1_message_proto = out.File

126 > file_temporal_server_api_metrics_v1_message_proto_goTypes = nil

127 > file_temporal_server_api_metrics_v1_message_proto_depIdxs = nil

128 }

go.temporal.io/server/api/namespace/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

114 }

115

116 > func init() { file_temporal_server_api_namespace_v1_message_proto_init() } message.pb.go

117 > func file_temporal_server_api_namespace_v1_message_proto_init() {

118 > if File_temporal_server_api_namespace_v1_message_proto != nil {

119 return

120 }

121 > type x struct{} message.pb.go

122 > out := protoimpl.TypeBuilder{

123 > File: protoimpl.DescBuilder{

124 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

125 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_namespace_v1_message_proto_rawDesc), len(file_temporal_server_api_namespace_v1_message_proto_rawDesc)),

126 > NumEnums: 0,

127 > NumMessages: 1,

128 > NumExtensions: 0,

129 > NumServices: 0,

130 > },

131 > GoTypes: file_temporal_server_api_namespace_v1_message_proto_goTypes,

132 > DependencyIndexes: file_temporal_server_api_namespace_v1_message_proto_depIdxs,

133 > MessageInfos: file_temporal_server_api_namespace_v1_message_proto_msgTypes,

134 > }.Build()

135 > File_temporal_server_api_namespace_v1_message_proto = out.File

136 > file_temporal_server_api_namespace_v1_message_proto_goTypes = nil

137 > file_temporal_server_api_namespace_v1_message_proto_depIdxs = nil

138 }

go.temporal.io/server/api/persistence/v1/chasm_visibility.pb.go 20 covered LOC · 2 ranges

Open complete file

146 }

147

148 > func init() { file_temporal_server_api_persistence_v1_chasm_visibility_proto_init() } chasm_visibility.pb.go

149 > func file_temporal_server_api_persistence_v1_chasm_visibility_proto_init() {

150 > if File_temporal_server_api_persistence_v1_chasm_visibility_proto != nil {

151 return

152 }

153 > type x struct{} chasm_visibility.pb.go

154 > out := protoimpl.TypeBuilder{

155 > File: protoimpl.DescBuilder{

156 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

157 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_chasm_visibility_proto_rawDesc), len(file_temporal_server_api_persistence_v1_chasm_visibility_proto_rawDesc)),

158 > NumEnums: 0,

159 > NumMessages: 2,

160 > NumExtensions: 0,

161 > NumServices: 0,

162 > },

163 > GoTypes: file_temporal_server_api_persistence_v1_chasm_visibility_proto_goTypes,

164 > DependencyIndexes: file_temporal_server_api_persistence_v1_chasm_visibility_proto_depIdxs,

165 > MessageInfos: file_temporal_server_api_persistence_v1_chasm_visibility_proto_msgTypes,

166 > }.Build()

167 > File_temporal_server_api_persistence_v1_chasm_visibility_proto = out.File

168 > file_temporal_server_api_persistence_v1_chasm_visibility_proto_goTypes = nil

169 > file_temporal_server_api_persistence_v1_chasm_visibility_proto_depIdxs = nil

170 }

go.temporal.io/server/api/persistence/v1/cluster_metadata.pb.go 20 covered LOC · 2 ranges

Open complete file

289 }

290

291 > func init() { file_temporal_server_api_persistence_v1_cluster_metadata_proto_init() } cluster_metadata.pb.go

292 > func file_temporal_server_api_persistence_v1_cluster_metadata_proto_init() {

293 > if File_temporal_server_api_persistence_v1_cluster_metadata_proto != nil {

294 return

295 }

296 > type x struct{} cluster_metadata.pb.go

297 > out := protoimpl.TypeBuilder{

298 > File: protoimpl.DescBuilder{

299 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

300 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_cluster_metadata_proto_rawDesc), len(file_temporal_server_api_persistence_v1_cluster_metadata_proto_rawDesc)),

301 > NumEnums: 0,

302 > NumMessages: 5,

303 > NumExtensions: 0,

304 > NumServices: 0,

305 > },

306 > GoTypes: file_temporal_server_api_persistence_v1_cluster_metadata_proto_goTypes,

307 > DependencyIndexes: file_temporal_server_api_persistence_v1_cluster_metadata_proto_depIdxs,

308 > MessageInfos: file_temporal_server_api_persistence_v1_cluster_metadata_proto_msgTypes,

309 > }.Build()

310 > File_temporal_server_api_persistence_v1_cluster_metadata_proto = out.File

311 > file_temporal_server_api_persistence_v1_cluster_metadata_proto_goTypes = nil

312 > file_temporal_server_api_persistence_v1_cluster_metadata_proto_depIdxs = nil

313 }

go.temporal.io/server/api/persistence/v1/history_tree.pb.go 20 covered LOC · 2 ranges

Open complete file

274 }

275

276 > func init() { file_temporal_server_api_persistence_v1_history_tree_proto_init() } history_tree.pb.go

277 > func file_temporal_server_api_persistence_v1_history_tree_proto_init() {

278 > if File_temporal_server_api_persistence_v1_history_tree_proto != nil {

279 return

280 }

281 > type x struct{} history_tree.pb.go

282 > out := protoimpl.TypeBuilder{

283 > File: protoimpl.DescBuilder{

284 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

285 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_history_tree_proto_rawDesc), len(file_temporal_server_api_persistence_v1_history_tree_proto_rawDesc)),

286 > NumEnums: 0,

287 > NumMessages: 3,

288 > NumExtensions: 0,

289 > NumServices: 0,

290 > },

291 > GoTypes: file_temporal_server_api_persistence_v1_history_tree_proto_goTypes,

292 > DependencyIndexes: file_temporal_server_api_persistence_v1_history_tree_proto_depIdxs,

293 > MessageInfos: file_temporal_server_api_persistence_v1_history_tree_proto_msgTypes,

294 > }.Build()

295 > File_temporal_server_api_persistence_v1_history_tree_proto = out.File

296 > file_temporal_server_api_persistence_v1_history_tree_proto_goTypes = nil

297 > file_temporal_server_api_persistence_v1_history_tree_proto_depIdxs = nil

298 }

go.temporal.io/server/api/persistence/v1/namespaces.pb.go 20 covered LOC · 2 ranges

Open complete file

536 }

537

538 > func init() { file_temporal_server_api_persistence_v1_namespaces_proto_init() } namespaces.pb.go

539 > func file_temporal_server_api_persistence_v1_namespaces_proto_init() {

540 > if File_temporal_server_api_persistence_v1_namespaces_proto != nil {

541 return

542 }

543 > type x struct{} namespaces.pb.go

544 > out := protoimpl.TypeBuilder{

545 > File: protoimpl.DescBuilder{

546 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

547 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_namespaces_proto_rawDesc), len(file_temporal_server_api_persistence_v1_namespaces_proto_rawDesc)),

548 > NumEnums: 0,

549 > NumMessages: 8,

550 > NumExtensions: 0,

551 > NumServices: 0,

552 > },

553 > GoTypes: file_temporal_server_api_persistence_v1_namespaces_proto_goTypes,

554 > DependencyIndexes: file_temporal_server_api_persistence_v1_namespaces_proto_depIdxs,

555 > MessageInfos: file_temporal_server_api_persistence_v1_namespaces_proto_msgTypes,

556 > }.Build()

557 > File_temporal_server_api_persistence_v1_namespaces_proto = out.File

558 > file_temporal_server_api_persistence_v1_namespaces_proto_goTypes = nil

559 > file_temporal_server_api_persistence_v1_namespaces_proto_depIdxs = nil

560 }

go.temporal.io/server/api/persistence/v1/queue_metadata.pb.go 20 covered LOC · 2 ranges

Open complete file

105 }

106

107 > func init() { file_temporal_server_api_persistence_v1_queue_metadata_proto_init() } queue_metadata.pb.go

108 > func file_temporal_server_api_persistence_v1_queue_metadata_proto_init() {

109 > if File_temporal_server_api_persistence_v1_queue_metadata_proto != nil {

110 return

111 }

112 > type x struct{} queue_metadata.pb.go

113 > out := protoimpl.TypeBuilder{

114 > File: protoimpl.DescBuilder{

115 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

116 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_queue_metadata_proto_rawDesc), len(file_temporal_server_api_persistence_v1_queue_metadata_proto_rawDesc)),

117 > NumEnums: 0,

118 > NumMessages: 2,

119 > NumExtensions: 0,

120 > NumServices: 0,

121 > },

122 > GoTypes: file_temporal_server_api_persistence_v1_queue_metadata_proto_goTypes,

123 > DependencyIndexes: file_temporal_server_api_persistence_v1_queue_metadata_proto_depIdxs,

124 > MessageInfos: file_temporal_server_api_persistence_v1_queue_metadata_proto_msgTypes,

125 > }.Build()

126 > File_temporal_server_api_persistence_v1_queue_metadata_proto = out.File

127 > file_temporal_server_api_persistence_v1_queue_metadata_proto_goTypes = nil

128 > file_temporal_server_api_persistence_v1_queue_metadata_proto_depIdxs = nil

129 }

go.temporal.io/server/api/persistence/v1/tasks.pb.go 20 covered LOC · 2 ranges

Open complete file

846 }

847

848 > func init() { file_temporal_server_api_persistence_v1_tasks_proto_init() } tasks.pb.go

849 > func file_temporal_server_api_persistence_v1_tasks_proto_init() {

850 > if File_temporal_server_api_persistence_v1_tasks_proto != nil {

851 return

852 }

853 > type x struct{} tasks.pb.go

854 > out := protoimpl.TypeBuilder{

855 > File: protoimpl.DescBuilder{

856 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

857 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_persistence_v1_tasks_proto_rawDesc), len(file_temporal_server_api_persistence_v1_tasks_proto_rawDesc)),

858 > NumEnums: 0,

859 > NumMessages: 8,

860 > NumExtensions: 0,

861 > NumServices: 0,

862 > },

863 > GoTypes: file_temporal_server_api_persistence_v1_tasks_proto_goTypes,

864 > DependencyIndexes: file_temporal_server_api_persistence_v1_tasks_proto_depIdxs,

865 > MessageInfos: file_temporal_server_api_persistence_v1_tasks_proto_msgTypes,

866 > }.Build()

867 > File_temporal_server_api_persistence_v1_tasks_proto = out.File

868 > file_temporal_server_api_persistence_v1_tasks_proto_goTypes = nil

869 > file_temporal_server_api_persistence_v1_tasks_proto_depIdxs = nil

870 }

go.temporal.io/server/api/token/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

785 }

786

787 > func init() { file_temporal_server_api_token_v1_message_proto_init() } message.pb.go

788 > func file_temporal_server_api_token_v1_message_proto_init() {

789 > if File_temporal_server_api_token_v1_message_proto != nil {

790 return

791 }

792 > type x struct{} message.pb.go

793 > out := protoimpl.TypeBuilder{

794 > File: protoimpl.DescBuilder{

795 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

796 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_token_v1_message_proto_rawDesc), len(file_temporal_server_api_token_v1_message_proto_rawDesc)),

797 > NumEnums: 0,

798 > NumMessages: 7,

799 > NumExtensions: 0,

800 > NumServices: 0,

801 > },

802 > GoTypes: file_temporal_server_api_token_v1_message_proto_goTypes,

803 > DependencyIndexes: file_temporal_server_api_token_v1_message_proto_depIdxs,

804 > MessageInfos: file_temporal_server_api_token_v1_message_proto_msgTypes,

805 > }.Build()

806 > File_temporal_server_api_token_v1_message_proto = out.File

807 > file_temporal_server_api_token_v1_message_proto_goTypes = nil

808 > file_temporal_server_api_token_v1_message_proto_depIdxs = nil

809 }

go.temporal.io/server/api/visibilityservice/v1/request_response.pb.go 20 covered LOC · 2 ranges

Open complete file

402 }

403

404 > func init() { file_temporal_server_api_visibilityservice_v1_request_response_proto_init() } request_response.pb.go

405 > func file_temporal_server_api_visibilityservice_v1_request_response_proto_init() {

406 > if File_temporal_server_api_visibilityservice_v1_request_response_proto != nil {

407 return

408 }

409 > type x struct{} request_response.pb.go

410 > out := protoimpl.TypeBuilder{

411 > File: protoimpl.DescBuilder{

412 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

413 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc), len(file_temporal_server_api_visibilityservice_v1_request_response_proto_rawDesc)),

414 > NumEnums: 0,

415 > NumMessages: 5,

416 > NumExtensions: 0,

417 > NumServices: 0,

418 > },

419 > GoTypes: file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes,

420 > DependencyIndexes: file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs,

421 > MessageInfos: file_temporal_server_api_visibilityservice_v1_request_response_proto_msgTypes,

422 > }.Build()

423 > File_temporal_server_api_visibilityservice_v1_request_response_proto = out.File

424 > file_temporal_server_api_visibilityservice_v1_request_response_proto_goTypes = nil

425 > file_temporal_server_api_visibilityservice_v1_request_response_proto_depIdxs = nil

426 }

go.temporal.io/server/api/workflow/v1/message.pb.go 20 covered LOC · 2 ranges

Open complete file

278 }

279

280 > func init() { file_temporal_server_api_workflow_v1_message_proto_init() } message.pb.go

281 > func file_temporal_server_api_workflow_v1_message_proto_init() {

282 > if File_temporal_server_api_workflow_v1_message_proto != nil {

283 return

284 }

285 > type x struct{} message.pb.go

286 > out := protoimpl.TypeBuilder{

287 > File: protoimpl.DescBuilder{

288 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

289 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_api_workflow_v1_message_proto_rawDesc), len(file_temporal_server_api_workflow_v1_message_proto_rawDesc)),

290 > NumEnums: 0,

291 > NumMessages: 3,

292 > NumExtensions: 0,

293 > NumServices: 0,

294 > },

295 > GoTypes: file_temporal_server_api_workflow_v1_message_proto_goTypes,

296 > DependencyIndexes: file_temporal_server_api_workflow_v1_message_proto_depIdxs,

297 > MessageInfos: file_temporal_server_api_workflow_v1_message_proto_msgTypes,

298 > }.Build()

299 > File_temporal_server_api_workflow_v1_message_proto = out.File

300 > file_temporal_server_api_workflow_v1_message_proto_goTypes = nil

301 > file_temporal_server_api_workflow_v1_message_proto_depIdxs = nil

302 }

go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1/service.pb.go 20 covered LOC · 2 ranges

Open complete file

86 }

87

88 > func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_init() } service.pb.go

89 > func file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_init() {

90 > if File_temporal_server_chasm_lib_scheduler_proto_v1_service_proto != nil {

91 return

92 }

93 > file_temporal_server_chasm_lib_scheduler_proto_v1_request_response_proto_init() service.pb.go

94 > type x struct{}

95 > out := protoimpl.TypeBuilder{

96 > File: protoimpl.DescBuilder{

97 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

98 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_rawDesc)),

99 > NumEnums: 0,

100 > NumMessages: 0,

101 > NumExtensions: 0,

102 > NumServices: 1,

103 > },

104 > GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_goTypes,

105 > DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_depIdxs,

106 > }.Build()

107 > File_temporal_server_chasm_lib_scheduler_proto_v1_service_proto = out.File

108 > file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_goTypes = nil

109 > file_temporal_server_chasm_lib_scheduler_proto_v1_service_proto_depIdxs = nil

110 }

go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1/tasks.pb.go 20 covered LOC · 2 ranges

Open complete file

343 }

344

345 > func init() { file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_init() } tasks.pb.go

346 > func file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_init() {

347 > if File_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto != nil {

348 return

349 }

350 > type x struct{} tasks.pb.go

351 > out := protoimpl.TypeBuilder{

352 > File: protoimpl.DescBuilder{

353 > GoPackagePath: reflect.TypeOf(x{}).PkgPath(),

354 > RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_rawDesc)),

355 > NumEnums: 0,

356 > NumMessages: 7,

357 > NumExtensions: 0,

358 > NumServices: 0,

359 > },

360 > GoTypes: file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_goTypes,

361 > DependencyIndexes: file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_depIdxs,

362 > MessageInfos: file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_msgTypes,

363 > }.Build()

364 > File_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto = out.File

365 > file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_goTypes = nil

366 > file_temporal_server_chasm_lib_scheduler_proto_v1_tasks_proto_depIdxs = nil

367 }

go.temporal.io/server/common/util/wildcard.go 20 covered LOC · 7 ranges

Open complete file

18 // WildCardStringToRegexps converts a given slices of string patterns to a slice of regular expressions matching

19 // wildcards (*) with any substring.

20 > func WildCardStringsToRegexp(patterns []string) (*regexp.Regexp, error) { wildcard.go

21 > var result strings.Builder

22 > result.WriteRune('^')

23 > for i, pattern := range patterns {

24 > result.WriteRune('(')

25 > first := true

26 > for literal := range strings.SplitSeq(pattern, "*") {

27 > if !first {

28 // Replace * with .*

29 result.WriteString(".*")

30 }

31 > result.WriteString(regexp.QuoteMeta(literal)) wildcard.go

32 > first = false

33 }

34 > result.WriteRune(')') wildcard.go

35 > if i < len(patterns)-1 {

36 > result.WriteRune('|') wildcard.go

37 > }

38 }

39 > result.WriteRune('$') wildcard.go

40 > return regexp.Compile(result.String())

41 }

42

43 // MustWildCardStringsToRegexp is like WildCardStringsToRegexp but panics on error.

44 > func MustWildCardStringsToRegexp(patterns []string) *regexp.Regexp { wildcard.go

45 > re, err := WildCardStringsToRegexp(patterns)

46 > if err != nil {

47 panic(err) //nolint:forbidigo // Must* functions conventionally panic on error.

48 }

49 > return re wildcard.go

50 }

go.temporal.io/server/common/dynamicconfig/collection.go 18 covered LOC · 3 ranges

Open complete file

676 // treat the fields independently), or the zero value of its type (if you want to treat the fields

677 // as a group and default unset fields to zero).

678 > func ConvertStructure[T any](def T) func(v any) (T, error) { collection.go

679 > return func(v any) (T, error) {

680 > // if we already have the right type, no conversion is necessary

681 > if typedV, ok := v.(T); ok {

682 return typedV, nil

683 }

685 // Deep-copy the default and decode over it. This allows using e.g. a struct with some

686 // default fields filled in and a config that only set some fields.

687 > out := deepCopyForMapstructure(def) collection.go

688 >

689 > dec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{

690 > Result: &out,

691 > DecodeHook: mapstructure.ComposeDecodeHookFunc(

692 > mapstructureHookDuration,

693 > mapstructureHookTimestamp,

694 > mapstructureHookProtoEnum,

695 > mapstructureHookGeneric,

696 > ),

697 > })

698 > if err != nil {

699 return out, err

700 }

701 > err = dec.Decode(v) collection.go

702 > return out, err

703 }

704 }

go.temporal.io/server/common/persistence/visibility/store/sql/query_converter_util_legacy.go 15 covered LOC · 3 ranges

Open complete file

68 }

69

70 > func newColName(name string) *colName { query_converter_util_legacy.go

71 > return &colName{Name: name}

72 > }

73

74 func newSAColName(

77 fieldName string,

78 valueType enumspb.IndexedValueType,

79 > ) *saColName { query_converter_util_legacy.go

80 > return &saColName{

81 > dbColName: newColName(dbColName),

82 > alias: alias,

83 > fieldName: fieldName,

84 > valueType: valueType,

85 > }

86 > }

87

88 func newFuncExpr(name string, exprs ...sqlparser.Expr) *sqlparser.FuncExpr {

105 }

106

107 > func getMaxDatetimeValue() time.Time { query_converter_util_legacy.go

108 > t, _ := time.Parse(time.RFC3339, "9999-12-31T23:59:59Z")

109 > return t

110 > }

111

112 // formatComparisonExprStringForError formats comparison expression after

go.temporal.io/server/common/build/build.go 14 covered LOC · 2 ranges

Open complete file

27 )

28

29 > func init() { build.go

30 > buildInfo, ok := debug.ReadBuildInfo()

31 > if !ok {

32 return

33 }

34

35 > InfoData.Available = true build.go

36 > InfoData.GoVersion = buildInfo.GoVersion

37 >

38 > for _, setting := range buildInfo.Settings {

39 > switch setting.Key {

40 > case "GOARCH":

41 > InfoData.GoArch = setting.Value

42 > case "GOOS":

43 > InfoData.GoOs = setting.Value

44 > case "CGO_ENABLED":

45 > InfoData.CgoEnabled = setting.Value == "1"

46 case "vcs.revision":

47 InfoData.GitRevision = setting.Value

go.temporal.io/server/common/dynamicconfig/gradual_change.go 14 covered LOC · 3 ranges

Open complete file

25 // StaticGradualChange returns a GradualChange whose Value always returns def and whose When

26 // always returns a time in the past.

27 > func StaticGradualChange[T any](def T) GradualChange[T] { gradual_change.go

28 > return GradualChange[T]{New: def}

29 > }

30

31 // Value returns the value for the given key at the given time.

56 // of type GradualChange into a GradualChange.

57 // nolint:revive // cognitive-complexity // this looks complicated but each case is fairly simple

58 > func ConvertGradualChange[T any](def T) func(v any) (GradualChange[T], error) { gradual_change.go

59 > changeConverter := ConvertStructure(StaticGradualChange(def))

60 >

61 > // Call this once so that if it's going to panic, it panics at static init time.

62 > _, _ = changeConverter(nil)

63 >

64 > switch reflect.TypeFor[T]() {

65 > case reflect.TypeFor[bool]():

66 > return func(v any) (GradualChange[T], error) {

67 if b, err := convertBool(v); err == nil {

68 var change GradualChange[T]

72 return changeConverter(v)

73 }

74 > case reflect.TypeFor[int](): gradual_change.go

75 > return func(v any) (GradualChange[T], error) {

76 if i, err := convertInt(v); err == nil {

77 var change GradualChange[T]

go.temporal.io/server/common/dynamicconfig/shared_structure.go 13 covered LOC · 5 ranges

Open complete file

17 )

18

19 > func warnDefaultSharedStructure(key string, def any) { shared_structure.go

20 > if path := hasSharedStructure(reflect.ValueOf(def), "root"); path != "" {

21 sharedStructureWarnings.Store(key, path)

22 }

42 }

43

44 > func hasSharedStructure(v reflect.Value, path string) string { shared_structure.go

45 > // nolint:exhaustive // deliberately not exhaustive

46 > switch v.Kind() {

47 > case reflect.Map, reflect.Slice, reflect.Pointer:

48 > if !v.IsNil() {

49 return path

50 }

51 > case reflect.Interface: shared_structure.go

52 > if !v.IsNil() {

53 return hasSharedStructure(v.Elem(), path)

54 }

55 > case reflect.Struct: shared_structure.go

56 > for i := range v.NumField() {

57 > if p := hasSharedStructure(v.Field(i), path+"."+v.Type().Field(i).Name); p != "" {

58 return p

59 }

66 }

67 }

68 > return "" shared_structure.go

69 }

go.temporal.io/server/common/metrics/defs_base.go 13 covered LOC · 2 ranges

Open complete file

10 }

11

12 > func newMetricDefinition(name string, opts ...Option) metricDefinition { defs_base.go

13 > d := metricDefinition{

14 > name: name,

15 > description: "",

16 > unit: "",

17 > }

18 > for _, opt := range opts {

19 > opt.apply(&d)

20 > }

21 > return d

22 }

23

24 > func (md metricDefinition) Name() string { defs_base.go

25 > return md.name

26 > }

27

28 func (md metricDefinition) Unit() MetricUnit {

go.temporal.io/server/common/persistence/visibility/store/elasticsearch/visibility_store.go 12 covered LOC · 3 ranges

Open complete file

101 }

102

103 > defaultSorter = func() []elastic.Sorter { visibility_store.go

104 > ret := make([]elastic.Sorter, 0, len(defaultSorterFields))

105 > for _, item := range defaultSorterFields {

106 > fs := elastic.NewFieldSort(item.name)

107 > if item.desc {

108 > fs.Desc()

109 > }

110 > if item.missing_first {

111 > fs.Missing("_first")

112 > } else {

113 fs.Missing("_last")

114 }

115 > ret = append(ret, fs) visibility_store.go

116 }

117 > return ret visibility_store.go

118 }()

119

go.temporal.io/server/common/persistence/sql/sqlplugin/postgresql/plugin.go 10 covered LOC · 1 range

Open complete file

36 var _ sqlplugin.Plugin = (*plugin)(nil)

37

38 > func init() { plugin.go

39 > sql.RegisterPlugin(PluginName, &plugin{

40 > driver: &driver.PQDriver{},

41 > queryConverter: &queryConverter{},

42 > })

43 > sql.RegisterPlugin(PluginNamePGX, &plugin{

44 > driver: &driver.PGXDriver{},

45 > queryConverter: &queryConverter{},

46 > })

47 > }

48

49 func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {

go.temporal.io/server/common/persistence/sql/sqlplugin/postgresql/visibility.go 10 covered LOC · 1 range

Open complete file

40 )

41

42 > func buildOnDuplicateKeyUpdate(fields ...string) string { visibility.go

43 > items := make([]string, len(fields))

44 > for i, field := range fields {

45 > items[i] = fmt.Sprintf("%s = excluded.%s", field, field)

46 > }

47 > return fmt.Sprintf(

48 > // The WHERE clause ensures that no update occurs if the version is behind the saved version.

49 > "ON CONFLICT (namespace_id, run_id) DO UPDATE SET %s WHERE executions_visibility.%s < EXCLUDED.%s",

50 > strings.Join(items, ", "), sqlplugin.VersionColumnName, sqlplugin.VersionColumnName,

51 > )

52 }

53

go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite/visibility.go 10 covered LOC · 1 range

Open complete file

42 )

43

44 > func buildOnDuplicateKeyUpdate(fields ...string) string { visibility.go

45 > items := make([]string, len(fields))

46 > for i, field := range fields {

47 > items[i] = fmt.Sprintf("%s = excluded.%s", field, field)

48 > }

49 > return fmt.Sprintf(

50 > // The WHERE clause ensures that no update occurs if the version is behind the saved version.

51 > "ON CONFLICT (namespace_id, run_id) DO UPDATE SET %s WHERE executions_visibility.%s < EXCLUDED.%s",

52 > strings.Join(items, ", "), sqlplugin.VersionColumnName, sqlplugin.VersionColumnName,

53 > )

54 }

55

go.temporal.io/server/common/persistence/sql/sqlplugin/visibility.go 10 covered LOC · 2 ranges

Open complete file

219 }

220

221 > func getDbFields() []string { visibility.go

222 > t := reflect.TypeFor[VisibilityRow]()

223 > dbFields := make([]string, t.NumField())

224 > for i := 0; i < t.NumField(); i++ {

225 > f := t.Field(i)

226 > dbFields[i] = f.Tag.Get("db")

227 > if dbFields[i] == "" {

228 > dbFields[i] = strcase.ToSnake(f.Name)

229 > }

230 }

231 > return dbFields visibility.go

232 }

233

go.temporal.io/server/common/persistence/visibility/store/query/util.go 10 covered LOC · 2 ranges

Open complete file

70 }

71

72 > func NewUnsafeSQLString(val string) *UnsafeSQLString { util.go

73 > return &UnsafeSQLString{Val: val}

74 > }

75

76 func NewColName(name string) *ColumnName {

78 }

79

80 > func NewSAColumn(alias string, fieldName string, valueType enumspb.IndexedValueType) *SAColumn { util.go

81 > return &SAColumn{

82 > Alias: alias,

83 > FieldName: fieldName,

84 > ValueType: valueType,

85 > }

86 > }

87

88 func NamespaceDivisionSAColumn() *SAColumn {

go.temporal.io/server/common/metrics/noop_impl.go 9 covered LOC · 5 ranges

Open complete file

15 )

16

17 > func newNoopMetricsHandler() *noopMetricsHandler { return &noopMetricsHandler{} } noop_impl.go

18

19 // WithTags creates a new MetricProvder with provided []Tag

29

30 // Gauge obtains a gauge for the given name.

31 > func (*noopMetricsHandler) Gauge(string) GaugeIface { noop_impl.go

32 > return NoopGaugeMetricFunc

33 > }

34

35 // Timer obtains a timer for the given name.

36 > func (*noopMetricsHandler) Timer(string) TimerIface { noop_impl.go

37 > return NoopTimerMetricFunc

38 > }

39

40 // Histogram obtains a histogram for the given name.

54

55 var NoopCounterMetricFunc = CounterFunc(func(i int64, t ...Tag) {})

56 > var NoopGaugeMetricFunc = GaugeFunc(func(f float64, t ...Tag) {}) noop_impl.go

57 > var NoopTimerMetricFunc = TimerFunc(func(d time.Duration, t ...Tag) {}) noop_impl.go

58 var NoopHistogramMetricFunc = HistogramFunc(func(i int64, t ...Tag) {})

go.temporal.io/server/common/persistence/sql/sqlplugin/util.go 9 covered LOC · 2 ranges

Open complete file

5 )

6

7 > func appendPrefix(prefix string, fields []string) []string { util.go

8 > out := make([]string, len(fields))

9 > for i, field := range fields {

10 > out[i] = prefix + field

11 > }

12 > return out

13 }

14

15 > func BuildNamedPlaceholder(fields ...string) string { util.go

16 > return strings.Join(appendPrefix(":", fields), ", ")

17 > }

go.temporal.io/server/common/testing/testhooks/test_impl.go 9 covered LOC · 2 ranges

Open complete file

89 var keyCounter atomic.Int64

90

91 > func newKey[T any, S any]() Key[T, S] { test_impl.go

92 > var zero S

93 > var s ScopeType

94 > switch any(zero).(type) {

95 > case namespace.ID, namespace.Name:

96 > s = ScopeNamespace

97 > case global:

98 > s = ScopeGlobal

99 default:

100 panic("testhooks: unknown scope type")

101 }

102 > return Key[T, S]{id: keyCounter.Add(1), scopeType: s} test_impl.go

103 }

go.temporal.io/server/common/persistence/sql/sqlplugin/mysql/visibility.go 8 covered LOC · 1 range

Open complete file

73 )

74

75 > func buildOnDuplicateKeyUpdate(fields ...string) string { visibility.go

76 > items := make([]string, len(fields))

77 > for i, field := range fields {

78 > // This line is to ensure that no update occurs (for any column) if the version is behind the saved version.

79 > items[i] = fmt.Sprintf("%v = IF(%v < VALUES(%v), VALUES(%v), %v)",

80 > field, sqlplugin.VersionColumnName, sqlplugin.VersionColumnName, field, field)

81 > }

82 > return fmt.Sprintf("ON DUPLICATE KEY UPDATE %s", strings.Join(items, ", "))

83 }

84

go.temporal.io/server/common/dynamicconfig/registry.go 7 covered LOC · 3 ranges

Open complete file

17 )

18

19 > func register(s GenericSetting) { registry.go

20 > if globalRegistry.queried.Load() {

21 panic("dynamicconfig.New*Setting must only be called from static initializers")

22 }

23 > if globalRegistry.settings == nil { registry.go

24 > globalRegistry.settings = make(map[Key]GenericSetting)

25 > }

26 > if globalRegistry.settings[s.Key()] != nil {

27 // nolint:forbidigo // only called during static initialization

28 panic(fmt.Sprintf("duplicate registration of dynamic config key: %q", s.Key().String()))

29 }

30 > globalRegistry.settings[s.Key()] = s registry.go

31 }

32

go.temporal.io/server/common/membership/grpc_resolver.go 7 covered LOC · 2 ranges

Open complete file

53 )

54

55 > func init() { grpc_resolver.go

56 > // This must be called in init to avoid race conditions.

57 > resolver.Register(&globalGrpcBuilder)

58 > }

59

60 // Most code should not use this, this is only exposed for code that has to recognize and use a

80 }

81

82 > func (m *grpcBuilder) Scheme() string { grpc_resolver.go

83 > return grpcResolverScheme

84 > }

85

86 func (m *grpcBuilder) getServiceResolver(u *url.URL) (ServiceResolver, error) {

go.temporal.io/server/common/searchattribute/name_type_map.go 7 covered LOC · 1 range

Open complete file

51

52 // NewNameTypeMap creates a new NameTypeMap with the given custom search attributes.

53 > func NewNameTypeMap(customSearchAttributes map[string]enumspb.IndexedValueType) NameTypeMap { name_type_map.go

54 > return NameTypeMap{

55 > systemSearchAttributes: system,

56 > predefinedSearchAttributes: predefined,

57 > customSearchAttributes: customSearchAttributes,

58 > }

59 > }

60

61 // WithSystemSearchAttributes returns a new NameTypeMap overriding the system search

go.temporal.io/server/common/util.go 7 covered LOC · 2 ranges

Open complete file

161

162 // CreatePersistenceClientRetryPolicy creates a retry policy for calls to persistence

163 > func CreatePersistenceClientRetryPolicy() backoff.RetryPolicy { util.go

164 > return backoff.NewExponentialRetryPolicy(persistenceClientRetryInitialInterval).

165 > WithMaximumAttempts(persistenceClientRetryMaxAttempts)

166 > }

167

168 // CreateFrontendClientRetryPolicy creates a retry policy for calls to frontend service

688

689 // CloneProto is a generic typed version of proto.Clone from proto.

690 > func CloneProto[T proto.Message](v T) T { util.go

691 > return proto.Clone(v).(T)

692 > }

693

694 func CloneProtoMap[K comparable, T proto.Message](src map[K]T) map[K]T {

go.temporal.io/server/common/clock/time_source.go 6 covered LOC · 2 ranges

Open complete file

31

32 // NewRealTimeSource returns a timeSource that uses the real wall timeSource time.

33 > func NewRealTimeSource() RealTimeSource { time_source.go

34 > return RealTimeSource{}

35 > }

36

37 // Now returns the current time, with the location set to UTC.

38 > func (ts RealTimeSource) Now() time.Time { time_source.go

39 > return time.Now().UTC()

40 > }

41

42 // Since returns the time elapsed since t

go.temporal.io/server/common/metrics/option.go 6 covered LOC · 2 ranges

Open complete file

10 type WithDescription string

11

12 > func (h WithDescription) apply(m *metricDefinition) { option.go

13 > m.description = string(h)

14 > }

15

16 // WithUnit sets the unit of a metric. See NewBytesHistogramDef for an example.

17 type WithUnit MetricUnit

18

19 > func (h WithUnit) apply(m *metricDefinition) { option.go

20 > m.unit = MetricUnit(h)

21 > }

go.temporal.io/server/common/payload/payload.go 6 covered LOC · 2 ranges

Open complete file

30 }

31

32 > func Encode(value any) (*commonpb.Payload, error) { payload.go

33 > return defaultDataConverter.ToPayload(value)

34 > }

35

36 > func Decode(p *commonpb.Payload, valuePtr any) error { payload.go

37 > return defaultDataConverter.FromPayload(p, valuePtr)

38 > }

39

40 func ToString(p *commonpb.Payload) string {

go.temporal.io/server/common/persistence/data_interfaces.go 6 covered LOC · 3 ranges

Open complete file

1408 // UnixMilliseconds returns t as a Unix time, the number of milliseconds elapsed since January 1, 1970 UTC.

1409 // It should be used for all CQL timestamp.

1410 > func UnixMilliseconds(t time.Time) int64 { data_interfaces.go

1411 > // Handling zero time separately because UnixNano is undefined for zero times.

1412 > if t.IsZero() {

1413 return 0

1414 }

1415

1416 > unixNano := t.UnixNano() data_interfaces.go

1417 > if unixNano < 0 {

1418 // Time is before January 1, 1970 UTC

1419 return 0

1420 }

1421 > return unixNano / int64(time.Millisecond) data_interfaces.go

1422 }

1423

go.temporal.io/server/common/persistence/nosql/nosqlplugin/cassandra/translator/fixed_address_translator.go 6 covered LOC · 2 ranges

Open complete file

15 )

16

17 > func init() { fixed_address_translator.go

18 > RegisterTranslator(fixedTranslatorName, NewFixedAddressTranslatorPlugin())

19 > }

20

21 type FixedAddressTranslatorPlugin struct {

22 }

23

24 > func NewFixedAddressTranslatorPlugin() TranslatorPlugin { fixed_address_translator.go

25 > return &FixedAddressTranslatorPlugin{}

26 > }

27

28 // GetTranslator What gocql driver does is that it will connect to the first node in the list in configuration

go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite/plugin.go 6 covered LOC · 1 range

Open complete file

43 }

44

45 > func init() { plugin.go

46 > sql.RegisterPlugin(PluginName, &plugin{

47 > queryConverter: &queryConverter{},

48 > connPool: newConnPool(),

49 > })

50 > }

51

52 func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {

go.temporal.io/server/service/history/tasks/key.go 6 covered LOC · 1 range

Open complete file

35 }

36

37 > func NewKey(fireTime time.Time, taskID int64) Key { key.go

38 > return Key{

39 > FireTime: fireTime,

40 > TaskID: taskID,

41 > }

42 > }

43

44 func ValidateKey(key Key) error {

go.temporal.io/server/common/metrics/registry.go 5 covered LOC · 1 range

Open complete file

43

44 // register adds a metric definition to the list of pending metric definitions. This method is thread-safe.

45 > func (c *registry) register(d metricDefinition) { registry.go

46 > c.Lock()

47 > defer c.Unlock()

48 > c.definitions = append(c.definitions, d)

49 > }

50

51 // buildCatalog builds a catalog from the list of pending metric definitions. It is safe to call this method multiple

go.temporal.io/server/common/persistence/sql/sqlplugin/mysql/plugin.go 5 covered LOC · 1 range

Open complete file

24 var _ sqlplugin.Plugin = (*plugin)(nil)

25

26 > func init() { plugin.go

27 > sql.RegisterPlugin(PluginName, &plugin{

28 > queryConverter: &queryConverter{},

29 > })

30 > }

31

32 func (p *plugin) GetVisibilityQueryConverter() sqlplugin.VisibilityQueryConverter {

go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite/conn_pool.go 5 covered LOC · 1 range

Open complete file

23 }

24

25 > func newConnPool() *connPool { conn_pool.go

26 > return &connPool{

27 > pool: make(map[string]entry),

28 > }

29 > }

30

31 // Allocate allocates the shared database in the pool or returns already exists instance with the same DSN. If instance

go.temporal.io/server/common/primitives/timestamp/duration.go 5 covered LOC · 3 ranges

Open complete file

19 )

20

21 > func DurationValue(d *durationpb.Duration) time.Duration { duration.go

22 > if d == nil {

23 > return 0 duration.go

24 > }

25 > return d.AsDuration() duration.go

26 }

27

go.temporal.io/server/common/log/with_logger.go 4 covered LOC · 2 ranges

Open complete file

14 // With returns Logger instance that prepend every log entry with tags. If logger implements

15 // WithLogger it is used, otherwise every log call will be intercepted.

16 > func With(logger Logger, tags ...tag.Tag) Logger { with_logger.go

17 > if wl, ok := logger.(WithLogger); ok {

18 > return wl.With(tags...) with_logger.go

19 > }

20 return newWithLogger(logger, tags...)

21 }

go.temporal.io/server/common/persistence/sql/sqlplugin/mysql/typeconv.go 4 covered LOC · 2 ranges

Open complete file

33 }

34

35 > func getMinMySQLDateTime() time.Time { typeconv.go

36 > t, err := time.Parse(time.RFC3339, "1000-01-01T00:00:00Z")

37 > if err != nil {

38 return time.Unix(0, 0).UTC()

39 }

40 > return t.UTC() typeconv.go

41 }

go.temporal.io/server/common/persistence/sql/sqlplugin/postgresql/typeconv.go 4 covered LOC · 2 ranges

Open complete file

35 }

36

37 > func getMinPostgreSQLDateTime() time.Time { typeconv.go

38 > t, err := time.Parse(time.RFC3339, "1000-01-01T00:00:00Z")

39 > if err != nil {

40 return time.Unix(0, 0).UTC()

41 }

42 > return t.UTC() typeconv.go

43 }

go.temporal.io/server/common/persistence/sql/sqlplugin/sqlite/typeconv.go 4 covered LOC · 2 ranges

Open complete file

33 }

34

35 > func getMinSQLiteDateTime() time.Time { typeconv.go

36 > t, err := time.Parse(time.RFC3339, "1000-01-01T00:00:00Z")

37 > if err != nil {

38 return time.Unix(0, 0).UTC()

39 }

40 > return t.UTC() typeconv.go

41 }

go.temporal.io/server/api/persistence/v1/predicates.go-helpers.pb.go 3 covered LOC · 1 range

Open complete file

17

18 // Size returns the size of the object, in bytes, once serialized

19 > func (val *Predicate) Size() int { predicates.go-helpers.pb.go

20 > return proto.Size(val)

21 > }

22

23 // Equal returns whether two Predicate values are equivalent by recursively

go.temporal.io/server/chasm/library.go 3 covered LOC · 1 range

Open complete file

56 // tasks within the CHASM framework.

57 // The format of the returned FQN is: "libName.name"

58 > func FullyQualifiedName(libName, name string) string { library.go

59 > return libName + "." + name

60 > }

go.temporal.io/server/chasm/registrable_component.go 3 covered LOC · 1 range

Open complete file

203 // The generated ID is used to uniquely identify components and tasks within the CHASM framework. The same FQN will

204 // always produce the same ID.

205 > func GenerateTypeID(fqn string) uint32 { registrable_component.go

206 > return farm.Fingerprint32([]byte(fqn))

207 > }

208

209 // hasBusinessIDAlias returns true if the component has a businessID alias configured

go.temporal.io/server/common/cache/size_getter.go 3 covered LOC · 2 ranges

Open complete file

14 )

15

16 > func getSize(value any) int { size_getter.go

17 > if v, ok := value.(SizeGetter); ok {

18 return v.CacheSize()

19 }

20 // if the object does not have a CacheSize() method, assume is count limit cache, which size should be 1

21 > return 1 size_getter.go

22 }

go.temporal.io/server/common/dynamicconfig/key.go 3 covered LOC · 1 range

Open complete file

13 )

14

15 > func MakeKey(s string) Key { key.go

16 > return Key{handle: unique.Make(strings.ToLower(s))}

17 > }

18

19 func (k Key) String() string {

go.temporal.io/server/common/persistence/nosql/nosqlplugin/cassandra/translator/translator_plugin.go 3 covered LOC · 1 range

Open complete file

22 // RegisterPlugin adds an auth plugin to the plugin registry

23 // it is only safe to use from a package init function

24 > func RegisterTranslator(name string, plugin TranslatorPlugin) { translator_plugin.go

25 > translators[name] = plugin

26 > }

27

28 func LookupTranslator(name string) (TranslatorPlugin, error) {

go.temporal.io/server/common/persistence/persistence_rate_limited_clients.go 3 covered LOC · 1 range

Open complete file

1182 baseAPI string,

1183 taskCategory tasks.Category,

1184 > ) string { persistence_rate_limited_clients.go

1185 > return baseAPI + taskCategory.Name()

1186 > }

go.temporal.io/server/common/persistence/sql/store.go 3 covered LOC · 2 ranges

Open complete file

19

20 // RegisterPlugin will register a SQL plugin

21 > func RegisterPlugin(pluginName string, plugin sqlplugin.Plugin) { store.go

22 > if _, ok := supportedPlugins[pluginName]; ok {

23 panic("plugin " + pluginName + " already registered")

24 }

25 > supportedPlugins[pluginName] = plugin store.go

26 }

27

go.temporal.io/server/common/primitives/timestamp/time.go 3 covered LOC · 2 ranges

Open complete file

11 }

12

13 > func TimeValue(t *timestamppb.Timestamp) time.Time { time.go

14 > if t == nil {

15 return time.Time{}

16 }

17 > return t.AsTime() time.go

18 }

19

go.temporal.io/server/common/tasks/priority.go 3 covered LOC · 1 range

Open complete file

77 func getPriority(

78 class, subClass Priority,

79 > ) Priority { priority.go

80 > return class | subClass

81 > }

go.temporal.io/server/service/history/tasks/category.go 3 covered LOC · 1 range

Open complete file

100 }

101

102 > func (c Category) Name() string { category.go

103 > return c.name

104 > }

105

106 func (c Category) Type() CategoryType {

go.temporal.io/server/common/metrics/metrics.go 2 covered LOC · 2 ranges

Open complete file

81

82 func (c CounterFunc) Record(v int64, tags ...Tag) { c(v, tags...) }

83 > func (c GaugeFunc) Record(v float64, tags ...Tag) { c(v, tags...) } metrics.go

84 > func (c TimerFunc) Record(v time.Duration, tags ...Tag) { c(v, tags...) } metrics.go

85 func (c HistogramFunc) Record(v int64, tags ...Tag) { c(v, tags...) }

go.temporal.io/server/common/persistence/client/fx.go 2 covered LOC · 1 range

Open complete file

224 }

225

226 > func managerProvider[T persistence.Closeable](newManagerFn func(Factory) (T, error)) func(Factory, fx.Lifecycle) (T, error) { fx.go

227 > return func(f Factory, lc fx.Lifecycle) (T, error) {

228 manager, err := newManagerFn(f) // passing receiver (Factory) as first argument.

229 if err != nil {

go.temporal.io/server/common/aggregate/noop_moving_window_average.go 1 covered LOC · 1 range

Open complete file

7 )

8

9 > func newNoopMovingWindowAverage() *noopMovingWindowAverage { return &noopMovingWindowAverage{} } noop_moving_window_average.go

10

11 func (a *noopMovingWindowAverage) Record(_ int64) {}

go.temporal.io/server/common/persistence/noop_health_signal_aggregator.go 1 covered LOC · 1 range

Open complete file

11 )

12

13 > func newNoopSignalAggregator() *noopSignalAggregator { return &noopSignalAggregator{} } noop_health_signal_aggregator.go

14

15 func (a *noopSignalAggregator) Start() {}