Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

queue

package

Go to main page

Versions in this module

v0

v0.1.41

Mar 21, 2026

Changes in this version

+ type ProgressWatchdog struct

+ func NewProgressWatchdog(parent context.Context, timeout time.Duration) (*ProgressWatchdog, context.Context)

+ func (w *ProgressWatchdog) Beat()

+ func (w *ProgressWatchdog) Stop()

type Queue

+ func (q *Queue) GetLastPullWasPartial() bool

+ func (q *Queue) GetWorkerCount() int

+ func (q *Queue) SetCopyResumeDstExistenceWindow(anchorPass, anchorRound int)

type QueueMode

+ const QueueModeCopyRetry

+ type QueueWatchdog struct

+ func NewQueueWatchdog(q *Queue, stallTimeout time.Duration) *QueueWatchdog

+ func (wd *QueueWatchdog) Beat()

+ func (wd *QueueWatchdog) Start()

+ func (wd *QueueWatchdog) Stop()

type SrcNodeMeta

+ TraversalStatus string

type TaskBase

+ CopyStatus string

+ LastError string

+ SrcTraversalStatus string

+ WorkerResult string

v0.1.4

Mar 21, 2026

Changes in this version

+ const TaskTypeCopy

+ const TaskTypeCopyFile

+ const TaskTypeCopyFolder

+ const TaskTypeDstTraversal

+ const TaskTypeSrcTraversal

+ const TaskTypeUpload

+ func BatchLoadExpectedChildrenByDSTIDs(database *db.DB, dstParentIDs []string, dstIDToPath map[string]string) (expectedFoldersMap map[string][]types.Folder, ...)

+ func BatchLoadRetryDstCleanup(database *db.DB, srcFolderIDs []string) (map[string]*RetryDstCleanup, error)

+ func BuildExpectedMapsFromDstWithChildren(dstBatch []db.FetchResult, childrenByDstID map[string][]*db.NodeState) (expectedFoldersMap map[string][]types.Folder, ...)

+ func SeedRootTask(queueType string, rootFolder types.Folder, rootNodeID string, database *db.DB) error

+ func SeedRootTaskWithSrcID(queueType string, rootFolder types.Folder, rootNodeID string, srcID string, ...) error

+ func SeedRootTasks(srcRoot types.Folder, dstRoot types.Folder, database *db.DB) error

+ type ChildResult struct

+ File types.File

+ Folder types.Folder

+ IsFile bool

+ SrcCopyStatus string

+ SrcID string

+ Status string

+ type CompletionCheckOptions struct

+ AdvanceRoundIfComplete bool

+ CheckFinalCompletion bool

+ CheckRoundComplete bool

+ WasFirstPull bool

+ type CopyTask struct

+ DstCtx types.ServiceContext

+ DstId string

+ SrcId string

+ type CopyWorker struct

+ func NewCopyWorker(id string, queue *Queue, database *db.DB, srcAdapter types.FSAdapter, ...) *CopyWorker

+ func (w *CopyWorker) Run()

+ type EngineCaches struct

+ Dst *NodeCache

+ Src *NodeCache

+ func NewEngineCaches() *EngineCaches

+ type ExternalQueueMetrics struct

+ Bytes int64

+ BytesPerSecond float64

+ DiscoveryRateItemsPerSec float64

+ Files int64

+ FilesDiscoveredTotal int64

+ Folders int64

+ FoldersDiscoveredTotal int64

+ ItemsPerSecond float64

+ Round int

+ Total int64

+ TotalDiscovered int64

+ TotalFailed int

+ TotalPending int

+ type InternalQueueMetrics struct

+ ActiveProcessingTime time.Duration

+ LastState QueueState

+ LastStateChangeTime time.Time

+ TasksCompletedWhileActive int64

+ TimeIdleNoWork time.Duration

+ TimePausedRoundBoundary time.Duration

+ TimeProcessing time.Duration

+ TimeRateLimited time.Duration

+ TimeWaitingOnFS time.Duration

+ TimeWaitingOnQueue time.Duration

+ WallClockTime time.Duration

+ type LevelCache struct

+ ByCopyStatus map[string]map[string]struct{}

+ ByParentPath map[string]map[string]*db.NodeState

+ ByPath map[string]*db.NodeState

+ ByTraversalStatus map[string]map[string]struct{}

+ Nodes map[string]*db.NodeState

+ func NewLevelCache() *LevelCache

+ func (lc *LevelCache) Clear()

+ func (lc *LevelCache) Count() int

+ func (lc *LevelCache) Get(id string) *db.NodeState

+ func (lc *LevelCache) GetByPath(path string) *db.NodeState

+ func (lc *LevelCache) GetRef(id string) *db.NodeState

+ func (lc *LevelCache) ListChildrenByParentPath(parentPath string) []*db.NodeState

+ func (lc *LevelCache) ListPending(afterID string, limit int) []*db.NodeState

+ func (lc *LevelCache) ListPendingCopy(afterID string, limit int, nodeType string) []*db.NodeState

+ func (lc *LevelCache) Put(id string, n *db.NodeState)

+ func (lc *LevelCache) PutIfAbsent(id string, n *db.NodeState) bool

+ func (lc *LevelCache) Snapshot() map[string]*db.NodeState

+ func (lc *LevelCache) UpdateStatus(id string, traversalStatus, copyStatus string)

+ type LevelStats struct

+ Completed int64

+ CopyFailed int64

+ CopyPending int64

+ CopySuccessful int64

+ Failed int64

+ Pending int64

+ Successful int64

+ type NodeCache struct

+ Levels map[int]*LevelCache

+ Stats map[int]*LevelStats

+ func NewNodeCache() *NodeCache

+ func (nc *NodeCache) DropLevel(depth int)

+ func (nc *NodeCache) EnsureLevel(depth int) *LevelCache

+ func (nc *NodeCache) EnsureLevelStats(depth int) *LevelStats

+ func (nc *NodeCache) GetLevel(depth int) *LevelCache

+ func (nc *NodeCache) GetLevelStats(depth int) *LevelStats

+ func (nc *NodeCache) IncrementCompleted(depth int)

+ func (nc *NodeCache) IncrementPending(depth int)

+ func (nc *NodeCache) LevelDepths() []int

+ func (nc *NodeCache) PromoteLevel(fromDepth, toDepth int)

+ func (nc *NodeCache) RecordCopyTransition(depth int, oldStatus, newStatus string)

+ func (nc *NodeCache) RecordTraversalTransition(depth int, oldStatus, newStatus string)

+ func (nc *NodeCache) SetLevelStats(depth int, pending, successful, failed, completed int64)

+ func (nc *NodeCache) TotalNodeCount() int

+ type Queue struct

+ func NewQueue(name string, maxRetries int, workerCount int, coordinator *QueueCoordinator) *Queue

+ func (q *Queue) Add(task *TaskBase) bool

+ func (q *Queue) AddWorker(worker Worker)

+ func (q *Queue) AdvanceCopyRound()

+ func (q *Queue) AdvanceTraversalRound()

+ func (q *Queue) CheckCopyCompletion(currentRound int, wasFirstPull bool) bool

+ func (q *Queue) CheckTraversalCompletion(currentRound int) bool

+ func (q *Queue) Clear()

+ func (q *Queue) Close()

+ func (q *Queue) CompleteCopyTask(task *TaskBase, executionDelta time.Duration)

+ func (q *Queue) CompleteTraversalTask(task *TaskBase, executionDelta time.Duration)

+ func (q *Queue) EnsureRoundExpectedFromStats()

+ func (q *Queue) FailCopyTask(task *TaskBase, executionDelta time.Duration)

+ func (q *Queue) FailTraversalTask(task *TaskBase, executionDelta time.Duration)

+ func (q *Queue) GetAverageExecutionTime() time.Duration

+ func (q *Queue) GetBytesTransferredTotal() int64

+ func (q *Queue) GetCopyPass() int

+ func (q *Queue) GetExecutionTimeBufferSize() int

+ func (q *Queue) GetExecutionTimeDeltas() []time.Duration

+ func (q *Queue) GetFilesCreatedTotal() int64

+ func (q *Queue) GetFilesDiscoveredTotal() int64

+ func (q *Queue) GetFoldersCreatedTotal() int64

+ func (q *Queue) GetFoldersDiscoveredTotal() int64

+ func (q *Queue) GetMode() QueueMode

+ func (q *Queue) GetPendingCount() int

+ func (q *Queue) GetRound() int

+ func (q *Queue) GetRoundStats(round int) *RoundStats

+ func (q *Queue) GetTotalCompleted() int

+ func (q *Queue) GetTotalDiscovered() int64

+ func (q *Queue) GetTotalFailed() int

+ func (q *Queue) InProgressCount() int

+ func (q *Queue) InitializeCopyWithContext(database *db.DB, srcAdapter, dstAdapter types.FSAdapter, ...)

+ func (q *Queue) InitializeWithContext(database *db.DB, adapter types.FSAdapter, shutdownCtx context.Context)

+ func (q *Queue) IsExhausted() bool

+ func (q *Queue) IsPaused() bool

+ func (q *Queue) Lease() *TaskBase

+ func (q *Queue) Name() string

+ func (q *Queue) NodeCache() *NodeCache

+ func (q *Queue) OtherNodeCache() *NodeCache

+ func (q *Queue) Pause()

+ func (q *Queue) PullCopyTasks(force bool)

+ func (q *Queue) PullRetryTasks(force bool)

+ func (q *Queue) PullTasksIfNeeded(force bool)

+ func (q *Queue) PullTraversalTasks(force bool)

+ func (q *Queue) RehydrateLevelFromDB(depth int)

+ func (q *Queue) ReportTaskResult(task *TaskBase, result TaskExecutionResult)

+ func (q *Queue) Resume()

+ func (q *Queue) Run()

+ func (q *Queue) SetCopyPass(pass int)

+ func (q *Queue) SetMaxKnownDepth(depth int)

+ func (q *Queue) SetMode(mode QueueMode)

+ func (q *Queue) SetNodeCache(c *NodeCache)

+ func (q *Queue) SetObserver(observer *QueueObserver)

+ func (q *Queue) SetOtherNodeCache(c *NodeCache)

+ func (q *Queue) SetRound(round int)

+ func (q *Queue) SetShutdownContext(ctx context.Context)

+ func (q *Queue) SetState(state QueueState)

+ func (q *Queue) SetStatsChannel(ch chan QueueStats)

+ func (q *Queue) SetTraversalCacheLoaded(loaded bool)

+ func (q *Queue) SetWorkers(workers []Worker)

+ func (q *Queue) Shutdown()

+ func (q *Queue) State() QueueState

+ func (q *Queue) Stats() QueueStats

+ func (q *Queue) TotalTracked() int

+ type QueueCoordinator struct

+ func NewQueueCoordinator() *QueueCoordinator

+ func (c *QueueCoordinator) CanDstStartRound(targetRound int) bool

+ func (c *QueueCoordinator) CanSrcStartRound(srcRound int) bool

+ func (c *QueueCoordinator) GetRound(which string) int

+ func (c *QueueCoordinator) IsCompleted(queueType string) bool

+ func (c *QueueCoordinator) MarkCompleted(which string)

+ func (c *QueueCoordinator) SetMaxSrcAhead(n int)

+ func (c *QueueCoordinator) UpdateRound(which string, round int)

+ func (c *QueueCoordinator) WaitSealBackpressure(round int, database *db.DB)

+ type QueueMode string

+ const QueueModeCopy

+ const QueueModeRetry

+ const QueueModeTraversal

+ type QueueObserver struct

+ func NewQueueObserver(database *db.DB, updateInterval time.Duration) *QueueObserver

+ func (o *QueueObserver) RegisterQueue(queueName string, queue *Queue)

+ func (o *QueueObserver) Start()

+ func (o *QueueObserver) Stop()

+ func (o *QueueObserver) UnregisterQueue(queueName string)

+ type QueueState string

+ const QueueStateCompleted

+ const QueueStatePaused

+ const QueueStateRunning

+ const QueueStateStopped

+ const QueueStateWaiting

+ type QueueStateSnapshot struct

+ Database *db.DB

+ FirstPullForRound bool

+ InProgressCount int

+ LastPullWasPartial bool

+ Mode QueueMode

+ PendingCount int

+ PullLowWM int

+ Pulling bool

+ Round int

+ State QueueState

+ type QueueStats struct

+ InProgress int

+ Name string

+ Pending int

+ Round int

+ TotalTracked int

+ Workers int

+ type RetryDstChild struct

+ Depth int

+ ID string

+ TraversalStatus string

+ type RetryDstCleanup struct

+ Children []RetryDstChild

+ DstDepth int

+ DstID string

+ DstOldStatus string

+ type RoundInfo struct

+ AvgTasksPerSec float64

+ ExpectedCount int

+ ItemsYielded int

+ LastPartialPull bool

+ LastPullTime time.Time

+ PullCount int

+ Round int

+ StartTime time.Time

+ TasksCompleted int

+ TasksFailed int

+ type RoundStats struct

+ Completed int

+ Expected int

+ Failed int

+ type SrcNodeMeta struct

+ CopyStatus string

+ Depth int

+ type TaskBase struct

+ Attempts int

+ BytesTransferred int64

+ CopyPass int

+ DiscoveredChildren []ChildResult

+ DstParentID string

+ ExpectedFiles []types.File

+ ExpectedFolders []types.Folder

+ ExpectedSrcIDMap map[string]string

+ ExpectedSrcNodeMeta map[string]SrcNodeMeta

+ File types.File

+ Folder types.Folder

+ ID string

+ LeaseTime time.Time

+ Locked bool

+ RetryDstCleanup *RetryDstCleanup

+ Round int

+ Status string

+ Type string

+ func (t *TaskBase) Identifier() string

+ func (t *TaskBase) IsFile() bool

+ func (t *TaskBase) IsFolder() bool

+ func (t *TaskBase) LocationPath() string

+ type TaskExecutionResult string

+ const TaskExecutionResultFailed

+ const TaskExecutionResultSuccessful

+ type TaskResult struct

+ Data any

+ Error error

+ Success bool

+ Task *TaskBase

+ type TraversalWorker struct

+ func NewTraversalWorker(id string, queue *Queue, database *db.DB, adapter types.FSAdapter, ...) *TraversalWorker

+ func (w *TraversalWorker) Run()

+ type UploadTask struct

+ DstCtx types.ServiceContext

+ DstId string

+ SrcId string

+ type Worker interface

+ Run func()