xact

README

Package xact is the top-level home for AIStore eXtended Actions (a.k.a. batch jobs): their static descriptor table, common interfaces, and shared lifecycle helpers.

An xaction is a long-running, asynchronous, batch job supported and tracked by AIS. Examples include global rebalance, resilver, LRU eviction, bucket copy, bucket transform, prefetch, download, EC encode, mirroring, list-objects, get-batch, and more.

Xaction kinds

As of 4.5, AIS defines 33 xaction kinds. The canonical source is xact.Table; the user-facing names are also visible via ais show job --help.

archive              blob-download  cleanup      copy-bucket  copy-objects  create-inventory
delete-objects       download       dsort        ec-bucket    ec-get        ec-put
ec-resp              elect-primary  etl-bucket   etl-inline   etl-objects   evict-objects
evict-remote-bucket  get-batch      index-shard  list         lru-eviction  mirror
prefetch-objects     promote-files  put-copies   rebalance    rechunk       rename-bucket
resilver             summary        warm-up-metadata

Subdirectories

• xreg - local xaction registry and lifecycle coordination.
• xs - concrete named xactions and shared implementations.
• other files in this package define common xaction descriptors, base structs, snapshots, statistics, notification helpers, and utility code.

For the canonical list of xaction kinds and their static properties, see xact.Table in xact/api_table.go.

Xaction kinds are generally the apc.Act* constants defined under api/apc.

Static descriptor table

xact.Table is the static source of truth for xaction kind properties.

Each entry maps an xaction kind to a Descriptor. The descriptor captures properties that would otherwise be scattered across the codebase:

• display name;
• access defaults;
• scope;
• whether the kind can be started through the generic start API;
• whether it changes cluster metadata;
• whether it refreshes capacity statistics;
• whether it is rebalance or resilver itself;
• whether it conflicts with rebalance or resilver;
• whether it must be aborted by rebalance;
• whether it can idle between requests;
• whether it exposes extended statistics;
• whether registry history should be kept briefly and quietly;
• whether it reports status to IC.

The table is intentionally static. Runtime state belongs to the local registry, xaction instances, snapshots, IC notifications, or action-specific managers.

Xaction scope

Each descriptor has a scope:

• ScopeG - cluster-wide xaction.
• ScopeB - bucket-scoped xaction.
• ScopeGB - either one bucket or all buckets.
• ScopeT - single-target xaction.

Scope describes the natural domain of the xaction kind. It does not by itself define how status is queried, how the xaction is started, or whether the xaction reports to IC.

For example, resilver is target-local and therefore ScopeT. Rebalance is cluster-wide. Many storage-service and multi-object operations are bucket-scoped.

Starting xactions

Descriptor.Startable means the xaction can be started via the generic StartXaction path.

A non-startable xaction may still be a normal xaction. It can have registry state, snapshots, statistics, abort semantics, and status reporting. It is simply started through a different path.

Examples of non-generic start paths include:

• bucket copy, move, and transform APIs;
• PUT-triggered EC and mirroring lifecycles;
• list-objects request handling;
• GetBatch request handling;
• action-specific managers such as dsort or download;
• periodic or resource-triggered background work.

Therefore, do not interpret Startable == false as "not a real xaction" or "not visible in status." It only excludes the generic start API.

Local registry

Each target maintains a local xaction registry.

The registry is responsible for local lifecycle coordination:

• create or renew a running xaction;
• find running xactions;
• enforce limited coexistence rules;
• abort selected xactions;
• keep recently finished xactions briefly for status/history;
• expose local snapshots.

The registry is local to each daemon. Cluster-wide status is built either by aggregating local snapshots from targets or by using IC status for xactions that report to IC.

Snapshots

A snapshot, or core.Snap, is a local description of an xaction instance.

Snapshots are the general-purpose status representation for xactions registered locally. A snapshot can include common fields such as kind, ID, bucket, start and end time, aborted/finished state, counters, and optional extended xaction-specific statistics.

The snaps-based model is the broadest status model: targets expose their local snapshots, and the caller or proxy aggregates them as needed.

This model is used by generic job display paths such as show job, and it is the fallback for xaction kinds that do not participate in generic IC status reporting.

Status and wait models

AIS has more than one status/wait model because xactions have different lifecycles and scaling requirements.

At a high level:

• snaps-based status queries local registry snapshots;
• IC status queries status reported to Information Center proxies;
• action-specific status uses custom machinery for xactions with specialized managers or protocols.

The descriptor table declares which generic model is valid for a given kind.

Snaps-based status

In the snaps-based model, each target exposes local xaction snapshots. A caller, usually via a proxy, aggregates per-target results.

This model is general and does not require targets to proactively report status to IC. It works for xactions that maintain local registry state, including kinds that are not IC-tracked.

Use snaps-based status/wait when:

• Descriptor.ICMode == ICNone;
• the xaction has local registry state but no generic IC status path;
• the caller needs target-by-target detail;
• a CLI or API path intentionally aggregates local target snapshots;
• the xaction has no meaningful IC terminal/progress reporting contract.

ICNone does not mean "no status." It means "no generic IC status path."

IC status

IC stands for Information Center.

IC is a bounded set of AIS proxies that includes the current primary. For xactions that report to IC, targets notify IC members about terminal state and, for some kinds, periodic progress. Clients can then query or wait on one IC proxy instead of polling every target.

This is primarily a scalability boundary.

IC bounds the client-facing query fanout to a single proxy regardless of cluster size. Target-side push notifications still travel from every target to each IC proxy, but the inbound rate is bounded by xaction lifecycle events, not by client poll frequency.

Generic IC status/wait is available only when the descriptor's ICMode is non-zero.

ICMode

Descriptor.ICMode declares whether and how the xaction kind reports status to IC.

Modes:

• ICNone - the kind does not support the generic IC status/wait path. Callers must use snaps-based status/wait or an action-specific status path.
• ICUponTerm - targets notify IC when the xaction reaches terminal state.
• ICUponProgress - targets periodically notify IC with progress.

ICUponProgress is normally used together with ICUponTerm: progress updates keep IC current while the xaction is running, and terminal notification closes the lifecycle.

Most IC-tracked xactions use ICUponTerm.

ActDownload is the current progress-reporting case: targets push periodic progress and also finalize on terminal state.

ICNone

ICNone can mean several different things depending on the xaction kind. The related reasons fall into three categories:

• intrinsic (e.g., ScopeT)
• action-specific machinery (e.g., dsort)
• historical scope where IC reporting could be added later (e.g., multi-object kinds that may idle between requests)

Again, ICNone does not mean "no state" or "no status." It only means that the generic IC status/wait APIs must not be used for that kind.

Waiting

Waiting is status plus a completion condition.

Generic wait APIs must use the correct status model:

• IC wait is valid for IC-tracked kinds. When the caller provides kind, the client validates ICMode before issuing the request.

  ID-only status/wait requests are also valid; in that case the server resolves the xaction ID and owns the final status/error semantics.

• Snaps-based wait is valid for all xactions. The completion condition is chosen by the caller: args.Finished() for normal lifecycles, args.Idle() for demand-driven or idle-cycle xactions where "finished" is delayed for a separately configured idle timeout after useful work is done (see xact.IdlesBeforeFinishing).

This distinction matters because a wait path that queries IC for a non-IC xaction can wait forever or return misleading state. Conversely, forcing every client to poll all targets for an IC-tracked xaction defeats the scalability benefit of IC.

When adding a new xaction kind, define its status/wait model explicitly in xact.Table.

Idling xactions

Some xactions are demand-driven and may temporarily have no work while still remaining alive. These kinds set Descriptor.Idles.

An idling xaction may transition through an idle state between requests. The absence of current work does not necessarily mean the xaction is permanently finished.

This affects status and IC reporting. For some idle-cycle xactions, a simple terminal notification model is not a good fit. Those kinds should either remain ICNone and use snaps-based or action-specific status, or explicitly implement progress-based IC reporting.

Examples of idle or demand-driven categories include:

• on-demand EC/mirroring lifecycles;
• multi-object operations;
• list-objects;
• GetBatch;
• long-running background managers with intermittent work.

Rebalance and resilver coexistence

Some xactions require a stable target set. Others can safely run concurrently with global rebalance or node-local resilver. The descriptor table captures this with two primary flags:

• ConflictRebRes
• AbortByReb

ConflictRebRes means the kind must not start while rebalance or resilver is already running. Conversely, rebalance or resilver must not start while this kind is running.

AbortByReb means an in-flight instance of this kind is aborted when a new global rebalance starts.

These flags often travel together: if a job needs a stable target set to start, it usually needs the same stable set to finish. Exceptions are deliberate and must be documented in api_table.go.

Current exception classes include:

• AbortByReb without ConflictRebRes: long-running or on-demand work that should not be refused merely because rebalance is currently running, but cannot correctly survive a topology change mid-flight.

• ConflictRebRes without AbortByReb: work where a partial result remains useful or recoverable, and aborting near completion would be wasteful.

Rebalance and resilver descriptors

Descriptor.Rebalance and Descriptor.Resilver mark the xactions that are themselves rebalance or resilver.

These flags are not generic conflict flags. They are used by registry and coexistence logic to recognize self-conflicts and rebalance/resilver-specific behavior.

ActMoveBck is rebalance-like and is marked accordingly because it moves data between nodes as part of a bucket move.

Stable target set

A stable target set is required when a job cannot correctly continue across a cluster membership change.

A kind requires a stable target set if it sets either:

• ConflictRebRes;
• AbortByReb.

Xactions that open intra-cluster peer-to-peer streams generally pin those streams to the Smap snapshot selected at construction. If the target set changes while such a job is running, the original routing and stream assumptions may no longer be valid.

Metadata and capacity effects

Descriptor.Metasync means the xaction changes cluster metadata and must participate in metadata synchronization semantics.

Descriptor.RefreshCap means capacity statistics should be refreshed on completion.

These flags are independent of status reporting. An xaction can update metadata or capacity while using IC status, snaps-based status, or an action-specific status path.

Extended statistics

Descriptor.ExtendedStats marks xactions that expose xaction-specific extended statistics in addition to common snapshot fields.

Extended statistics belong to the snapshot/status representation. They do not imply a particular status transport. They may be surfaced through snaps-based queries, action-specific paths, or IC-backed status depending on the kind.

Quiet brief history

Descriptor.QuietBrief suppresses verbose per-state log records and keeps registry history only briefly.

This is useful for high-churn or request-driven xactions where long registry history would create noise without much operational value.

Action-specific status paths

Some xactions intentionally do not use the generic IC status path because they have their own manager, protocol, or aggregation model.

Examples include:

• dsort, which has its own manager and notification machinery;
• list-objects, where clients stream pages directly and generic job display uses snapshots;
• GetBatch/x-moss, which is proxy-coordinated and uses target-direct status;
• synchronous operations where a proxy aggregates per-target results and returns directly to the caller.

For these kinds, keep ICMode == ICNone and document the intended status path near the descriptor entry.

Adding a new xaction kind

When adding a new xaction kind, update xact.Table first.

Decide and document:

1. Scope:

   • cluster, bucket, bucket-or-global, or target-local.

2. Start path:

   • generic StartXaction, action-specific API, workload-triggered, periodic, or internal.

3. Registry behavior:

   • whether the xaction registers locally;
   • whether it can idle;
   • how long finished state should remain visible.

4. Status/wait model:

   • IC status/wait;
   • snaps-based status/wait;
   • action-specific status/wait.

5. IC reporting:

   • ICNone;
   • ICUponTerm;
   • ICUponProgress;
   • or a combination.

6. Rebalance/resilver coexistence:

   • whether a stable target set is required;
   • whether to refuse start during rebalance/resilver;
   • whether to abort on new rebalance.

7. Metadata and capacity effects:

   • Metasync;
   • RefreshCap.

8. Extended statistics:

   • whether common snapshot fields are enough;
   • whether Snap.Ext should carry action-specific data.

9. User-facing behavior:

   • CLI display;
   • wait semantics;
   • abort semantics;
   • documentation links.

If the kind has ICMode == ICNone, make sure generic IC status/wait callers fail explicitly or route to the correct snaps-based or action-specific path.

Common mistakes

Avoid treating Startable == false as "not visible" or "not a real xaction."

Avoid treating ICNone as "no status."

Avoid using generic IC wait for a kind that does not report to IC.

Avoid adding a new xaction kind without deciding how it behaves during global rebalance and node-local resilver.

Avoid assuming that idle means finished.

Avoid adding action-specific status machinery without documenting why generic snaps or IC status is insufficient.

User-facing documentation

This README is developer-facing. It describes internal xaction structure and status models.

For user-facing commands, examples, and operational documentation, see:

• docs/cli/job.md
• docs/cli/dsort.md
• docs/cli/download.md
• docs/rebalance.md
• docs/batch.md
• docs/cli/object.md
• docs/storage_svcs.md

Documentation

Overview

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2025-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.

Package xact provides core functionality for the AIStore eXtended Actions (xactions).

• Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.

Index

• Constants
• Variables
• func CheckValidKind(kind string) (err error)
• func CheckValidKindIC(kind string) (err error)
• func CheckValidUUID(id string) (err error)
• func Cname(kind, uuid string) string
• func CompareRebIDs(someID, fltID string) int
• func GetCtxVlabs(ctx context.Context) map[string]string
• func GetKindName(kindOrName string) (kind, name string)
• func GetSimilar(kindOrName string) (simKind, simName string)
• func GoRunW(xctn core.Xact)
• func IdlesBeforeFinishing(kindOrName string) bool
• func IsErrRecvAbortWI(err error) bool
• func IsErrRecvAbortXact(err error) bool
• func IsSameScope(kindOrName string, scs ...int) bool
• func IsValidKind(kind string) bool
• func IsValidRebID(id string) (valid bool)
• func IsValidUUID(id string) bool
• func ListDisplayNames(onlyStartable bool) (names []string)
• func NewCtxVlabs(vlabs map[string]string) context.Context
• func ParseCname(cname string) (xactKind, xactID string, _ error)
• func RebID2S(id int64) string
• func RefcntQuiCB(refc *atomic.Int32, maxTimeout, totalSoFar time.Duration) core.QuiRes
• func S2RebID(id string) (int64, error)
• func TuneNumWorkers(xname string, numWorkers, numMpaths int) (int, error)
• type ArgsMsg
  • func (args *ArgsMsg) Finished() SnapsCond
  • func (args *ArgsMsg) Idle() SnapsCond
  • func (args *ArgsMsg) NotRunning() SnapsCond
  • func (args *ArgsMsg) Started() SnapsCond
  • func (args *ArgsMsg) String() string
• type Base
  • func (xctn *Base) Abort(err error) bool
  • func (xctn *Base) AbortErr() error
  • func (xctn *Base) AbortedAfter(d time.Duration) (err error)
  • func (xctn *Base) AddErr(err error, logExtra ...int)
  • func (xctn *Base) AddNotif(n core.Notif)
  • func (xctn *Base) BcastCtrl(smap *meta.Smap, wid, opcode string, err error)
  • func (xctn *Base) Bck() *meta.Bck
  • func (xctn *Base) Bytes() int64
  • func (xctn *Base) ChanAbort() <-chan error
  • func (xctn *Base) Cname() string
  • func (xctn *Base) EndTime() time.Time
  • func (xctn *Base) Err() (err error)
  • func (xctn *Base) ErrCnt() int
  • func (xctn *Base) Finish()
  • func (*Base) FromTo() (*meta.Bck, *meta.Bck)
  • func (xctn *Base) ID() string
  • func (xctn *Base) InBytes() int64
  • func (xctn *Base) InObjs() int64
  • func (xctn *Base) InObjsAdd(cnt int, size int64)
  • func (xctn *Base) InitBase(id, kind string, bck *meta.Bck)
  • func (xctn *Base) IsAborted() bool
  • func (xctn *Base) IsDone() bool
  • func (xctn *Base) IsIdle() bool
  • func (xctn *Base) IsRunning() (yes bool)
  • func (xctn *Base) JoinErr() (int, error)
  • func (xctn *Base) Kind() string
  • func (xctn *Base) LomAdd(lom *core.LOM)
  • func (xctn *Base) Name() string
  • func (xctn *Base) NewErrRecvAbortWI(tid, wid, errCause string) error
  • func (xctn *Base) NewErrRecvAbortXact(tid, errCause string) error
  • func (xctn *Base) NewSnap(self core.Xact) (snap *core.Snap)
  • func (xctn *Base) Objs() int64
  • func (xctn *Base) ObjsAdd(cnt int, size int64)
  • func (xctn *Base) OutBytes() int64
  • func (xctn *Base) OutObjs() int64
  • func (xctn *Base) OutObjsAdd(cnt int, size int64)
  • func (xctn *Base) Quiesce(d time.Duration, cb core.QuiCB) core.QuiRes
  • func (xctn *Base) SendCtrl(tsi *meta.Snode, wid, opcode string, body []byte) error
  • func (xctn *Base) SetStopping() bool
  • func (xctn *Base) StartTime() time.Time
  • func (xctn *Base) String() string
  • func (xctn *Base) ToStats(stats *core.Stats)
• type BckJog
  • func (r *BckJog) Init(id, kind string, bck *meta.Bck, opts *mpather.JgroupOpts, config *cmn.Config)
  • func (r *BckJog) NumJoggers() int
  • func (r *BckJog) NumVisits() int64
  • func (r *BckJog) Run()
  • func (r *BckJog) Wait() error
• type BckJogRunner
  • func (r *BckJogRunner) Init(id, kind string, bck *meta.Bck, opts BckJogRunnerOpts, config *cmn.Config) error
  • func (r *BckJogRunner) NumWorkers() int
  • func (r *BckJogRunner) Run()
  • func (r *BckJogRunner) Wait() error
  • func (r *BckJogRunner) WorkChanFull() int64
• type BckJogRunnerOpts
• type Demand
• type DemandBase
  • func (r *DemandBase) Abort(err error) (ok bool)
  • func (r *DemandBase) DecPending()
  • func (r *DemandBase) IdleTimer() <-chan struct{}
  • func (r *DemandBase) IncPending()
  • func (r *DemandBase) Init(uuid, kind string, bck *meta.Bck, idleDur time.Duration, hkcb ...hk.HKCB)
  • func (r *DemandBase) IsIdle() bool
  • func (r *DemandBase) Pending() (cnt int64)
  • func (r *DemandBase) Reset(idleTime time.Duration)
  • func (r *DemandBase) Stop()
  • func (r *DemandBase) SubPending(n int)
• type Descriptor
  • func GetDescriptor(kindOrName string) (string, Descriptor, error)
• type ICMode
• type Marked
• type MultiSnap
  • func (xs MultiSnap) AggregateState(xid string) (aborted, running, notstarted bool)
  • func (xs MultiSnap) ByteCounts(xid string) (locBytes, outBytes, inBytes int64)
  • func (xs MultiSnap) GetUUIDs() []string
  • func (xs MultiSnap) ObjCounts(xid string) (locObjs, outObjs, inObjs int64)
  • func (xs MultiSnap) RunningTarget(xid string) (string, *core.Snap, error)
  • func (xs MultiSnap) ToJSON(tid string, indent bool) ([]byte, error)
  • func (xs MultiSnap) TotalRunningTime(xid string) (time.Duration, error)
• type NotifXact
  • func (nx *NotifXact) ToNotifMsg(aborted bool) core.NotifMsg
• type NotifXactListener
  • func NewXactNL(uuid, kind string, smap *meta.Smap, srcs meta.NodeMap, bck ...*cmn.Bck) *NotifXactListener
  • func (nxb *NotifXactListener) QueryArgs() cmn.HreqArgs
  • func (*NotifXactListener) UnmarshalStats(rawMsg []byte) (stats any, finished, aborted bool, err error)
  • func (nxb *NotifXactListener) WithCause(cause string) *NotifXactListener
• type QueryMsg
  • func (msg *QueryMsg) String() (s string)
• type SnapsCond

Constants

const (
	PrefixDnlID = "dnl-" // http downloader (not to confuse with blob-downloader)
	PrefixEtlID = "etl-" // ETL
	PrefixTcoID = "tco-" // transform/copy objects
	PrefixInvID = "inv-" // create bucket inventory (NBI)
	PrefixGbtID = "gbt-" // get-batch (internally, x-moss)
	PrefixSrtID = "srt-" // conditional linkage (build tag "dsort")

	// evict
	PrefixEvictKeepID   = "kpmd-"
	PrefixEvictRemoveID = "rmmd-"
)

const (
	SepaID = ","

	LeftID  = "["
	RightID = "]"
)

const (
	ScopeG  = iota + 1 // cluster
	ScopeB             // bucket
	ScopeGB            // (one bucket) | (all buckets)
	ScopeT             // target
)

const (
	DefWaitTimeShort = time.Minute        // zero `ArgsMsg.Timeout` defaults to
	DefWaitTimeLong  = 7 * 24 * time.Hour // when `ArgsMsg.Timeout` is negative
	MaxProbingFreq   = 30 * time.Second   // as the name implies
	MinPollTime      = 2 * time.Second    // ditto
	MaxPollTime      = 2 * time.Minute    // can grow up to

)

global waiting tunables

const (
	FlagZeroSize = 1 << iota // usage: x-cleanup (apc.ActStoreCleanup) to remove zero size objects
	FlagLatestVer
	FlagSync

	// the flag overrides the default space-cleanup job operation making it NOT to remove (ie, keep)
	// misplaced objects
	FlagKeepMisplaced

	// makes global rebalance run in special cleanup mode,
	// safely removing misplaced objects
	FlagRemoveMisplaced
)

ArgsMsg.Flags note: for simplicity, keeping all custom x-flags in one place and one global enum for now

const (
	NwpBurstMult = 48   // channel size = burst * num-workers
	NwpBurstMax  = 8192 // upper bound on (shared) workCh size
)

shared work channel sizing

const (
	NwpNone = -1 // no workers: iterated LOMs executed by the iterating goroutine
	NwpMin  = 2  // throttled minimum
	NwpDflt = 0  // resolve at runtime via defaultNW()
)

num-workers specials

const (
	OpcodeStartedWI = "startedWI"
	OpcodeAbortXact = "abortXact"
	OpcodeAbortWI   = "abortWI"
)

const (
	IdleDefault = time.Minute // hk -> idle tick
)

const NoneWID = "-" // wid is optional, may not be present

const NumT2TCtrlItems = 5

const (
	T2TCtrl = "t2tctrl" // see also: proxyToContTCO
)

POST /v1/xactions/{one of the constants below}

Variables

var IncFinished func()

var Table = map[string]Descriptor{

	apc.ActElection:  {DisplayName: "elect-primary", Scope: ScopeG, Startable: false},
	apc.ActRebalance: {Scope: ScopeG, Startable: true, Metasync: true, Rebalance: true, ICMode: ICUponTerm},

	apc.ActETLInline: {Scope: ScopeG, Startable: false, AbortByReb: true, ICMode: ICUponTerm},

	apc.ActLRU:          {DisplayName: "lru-eviction", Scope: ScopeGB, Startable: true, ICMode: ICUponTerm},
	apc.ActStoreCleanup: {DisplayName: "cleanup", Scope: ScopeGB, Startable: true, ConflictRebRes: true, ICMode: ICUponTerm},

	apc.ActSummaryBck: {
		DisplayName: "summary",
		Scope:       ScopeGB,
		Access:      apc.AceObjLIST | apc.AceBckHEAD,
		Startable:   false,
		Metasync:    false,
	},
	apc.ActSummaryShard: {
		DisplayName: "shard-summary",
		Scope:       ScopeB,
		Access:      apc.AceObjLIST,
		Startable:   false,
		Metasync:    false,
	},

	apc.ActResilver: {Scope: ScopeT, Startable: true, Resilver: true},
	apc.ActRechunk:  {Scope: ScopeB, Startable: true, RefreshCap: true, ConflictRebRes: true, AbortByReb: true, ICMode: ICUponTerm},

	apc.ActIndexShard: {Scope: ScopeB, Startable: true, RefreshCap: false, ConflictRebRes: true, AbortByReb: false, ICMode: ICUponTerm},

	apc.ActECGet:     {Scope: ScopeB, Startable: false, Idles: true, ExtendedStats: true},
	apc.ActECPut:     {Scope: ScopeB, Startable: false, RefreshCap: true, Idles: true, ExtendedStats: true},
	apc.ActECRespond: {Scope: ScopeB, Startable: false, Idles: true},
	apc.ActPutCopies: {Scope: ScopeB, Startable: false, RefreshCap: true, Idles: true},

	apc.ActArchive: {Scope: ScopeB, Access: apc.AccessRW, Startable: false, RefreshCap: true, Idles: true},

	apc.ActCopyObjects: {
		DisplayName:    "copy-objects",
		Scope:          ScopeB,
		Access:         apc.AccessRW,
		Startable:      false,
		RefreshCap:     true,
		Idles:          true,
		ConflictRebRes: true,
		AbortByReb:     true,
	},

	apc.ActETLObjects: {
		DisplayName:    "etl-objects",
		Scope:          ScopeB,
		Access:         apc.AccessRW,
		Startable:      false,
		RefreshCap:     true,
		Idles:          true,
		ConflictRebRes: true,
		AbortByReb:     true,
	},

	apc.ActBlobDl: {Access: apc.AccessRW, Scope: ScopeB, Startable: true, AbortByReb: true, RefreshCap: true, ICMode: ICUponTerm},

	apc.ActDownload: {Access: apc.AccessRW, Scope: ScopeG, Startable: false, Idles: true, AbortByReb: true, ICMode: ICUponTerm | ICUponProgress},

	apc.ActDsort: {
		DisplayName:    "dsort",
		Scope:          ScopeB,
		Access:         apc.AccessRW,
		Startable:      false,
		RefreshCap:     true,
		ExtendedStats:  true,
		ConflictRebRes: true,
		AbortByReb:     true,
	},

	apc.ActPromote: {
		DisplayName: "promote-files",
		Scope:       ScopeB,
		Access:      apc.AcePromote,
		Startable:   false,
		RefreshCap:  true,
		ICMode:      ICUponTerm,
	},
	apc.ActEvictObjects: {
		DisplayName: "evict-objects",
		Scope:       ScopeB,
		Access:      apc.AceObjDELETE,
		Startable:   false,
		RefreshCap:  true,
		ICMode:      ICUponTerm,
	},
	apc.ActEvictRemoteBck: {
		DisplayName: "evict-remote-bucket",
		Scope:       ScopeB,
		Access:      apc.AceObjDELETE,
		Startable:   false,
		RefreshCap:  true,
	},
	apc.ActDeleteObjects: {
		DisplayName: "delete-objects",
		Scope:       ScopeB,
		Access:      apc.AceObjDELETE,
		Startable:   false,
		RefreshCap:  true,
		ICMode:      ICUponTerm,
	},

	apc.ActPrefetchObjects: {
		DisplayName: "prefetch-objects",
		Scope:       ScopeB,
		Access:      apc.AccessRW,
		Startable:   true,
		RefreshCap:  true,
		ICMode:      ICUponTerm,
	},

	apc.ActECEncode: {
		DisplayName:    "ec-bucket",
		Scope:          ScopeB,
		Access:         apc.AccessRW,
		Startable:      true,
		Metasync:       true,
		RefreshCap:     true,
		ConflictRebRes: true,
		AbortByReb:     true,
		ICMode:         ICUponTerm,
	},
	apc.ActMakeNCopies: {
		DisplayName: "mirror",
		Scope:       ScopeB,
		Access:      apc.AccessRW,
		Startable:   true,
		Metasync:    true,
		RefreshCap:  true,
		ICMode:      ICUponTerm,
	},
	apc.ActMoveBck: {
		DisplayName:    "rename-bucket",
		Scope:          ScopeB,
		Access:         apc.AceMoveBucket,
		Startable:      false,
		Metasync:       true,
		Rebalance:      true,
		ConflictRebRes: true,
		AbortByReb:     true,
		ICMode:         ICUponTerm,
	},
	apc.ActCopyBck: {
		DisplayName:    "copy-bucket",
		Scope:          ScopeB,
		Access:         apc.AccessRW,
		Startable:      false,
		Metasync:       true,
		RefreshCap:     true,
		ConflictRebRes: true,
		AbortByReb:     true,
		ICMode:         ICUponTerm,
	},
	apc.ActETLBck: {
		DisplayName:    "etl-bucket",
		Scope:          ScopeB,
		Access:         apc.AccessRW,
		Startable:      false,
		Metasync:       true,
		RefreshCap:     true,
		ConflictRebRes: true,
		AbortByReb:     true,
		ICMode:         ICUponTerm,
	},

	apc.ActList: {Scope: ScopeB, Access: apc.AceObjLIST, Startable: false, Metasync: false, Idles: true, QuietBrief: true, ICMode: ICNone},

	apc.ActGetBatch: {Scope: ScopeGB, Startable: false, Metasync: false, ConflictRebRes: true, AbortByReb: true, Idles: true, QuietBrief: true},

	apc.ActCreateNBI: {Scope: ScopeB, Startable: false, Metasync: false, ConflictRebRes: true, AbortByReb: true, Idles: false, ICMode: ICUponTerm},

	apc.ActLoadLomCache: {DisplayName: "warm-up-metadata", Scope: ScopeB, Startable: true},
}

Functions

func CheckValidKind

func CheckValidKind(kind string) (err error)

func CheckValidKindIC

func CheckValidKindIC(kind string) (err error)

func CheckValidUUID

func CheckValidUUID(id string) (err error)

func Cname

func Cname(kind, uuid string) string

func CompareRebIDs

func CompareRebIDs(someID, fltID string) int

func GetCtxVlabs

func GetCtxVlabs(ctx context.Context) map[string]string

func GetKindName

func GetKindName(kindOrName string) (kind, name string)

func GetSimilar

func GetSimilar(kindOrName string) (simKind, simName string)

func GoRunW

func GoRunW(xctn core.Xact)

common helper to go-run and wait until it actually starts running

func IdlesBeforeFinishing

func IdlesBeforeFinishing(kindOrName string) bool

func IsErrRecvAbortWI

func IsErrRecvAbortWI(err error) bool

func IsErrRecvAbortXact

func IsErrRecvAbortXact(err error) bool

func IsSameScope

func IsSameScope(kindOrName string, scs ...int) bool

func IsValidKind

func IsValidKind(kind string) bool

func IsValidRebID

func IsValidRebID(id string) (valid bool)

func IsValidUUID

func IsValidUUID(id string) bool

func ListDisplayNames

func ListDisplayNames(onlyStartable bool) (names []string)

func NewCtxVlabs

func NewCtxVlabs(vlabs map[string]string) context.Context

func ParseCname

func ParseCname(cname string) (xactKind, xactID string, _ error)

func RebID2S

func RebID2S(id int64) string

func RefcntQuiCB

func RefcntQuiCB(refc *atomic.Int32, maxTimeout, totalSoFar time.Duration) core.QuiRes

common ref-counted quiescence

func S2RebID

func S2RebID(id string) (int64, error)

func TuneNumWorkers

func TuneNumWorkers(xname string, numWorkers, numMpaths int) (int, error)

estimate the requested number of workers given: a) user-specified number, if any b) current load, and c) media type (NVMe, etc.)

used by all list-range type jobs, tcb, and blob-download - xname: xaction name (for logging) - numWorkers (when non-zero): requested number of workers - numMpaths: number of available mountpaths

Types

type ArgsMsg

type ArgsMsg struct {
	ID          string        // xaction UUID
	Kind        string        // xaction kind _or_ name (see `xact.Table`)
	DaemonID    string        // node that runs this xaction
	Bck         cmn.Bck       // bucket
	Buckets     []cmn.Bck     // list of buckets (e.g., copy-bucket, lru-evict, etc.)
	Timeout     time.Duration // max time to wait
	Flags       uint32        `json:"flags,omitempty"` // enum (FlagZeroSize, ...) bitwise
	Force       bool          // force
	OnlyRunning bool          // only for running xactions
}

either xaction ID or Kind must be specified is getting passed via ActMsg.Value w/ MorphMarshal extraction

func (*ArgsMsg) Finished

func (args *ArgsMsg) Finished() SnapsCond

func (*ArgsMsg) Idle

func (args *ArgsMsg) Idle() SnapsCond

func (*ArgsMsg) NotRunning

func (args *ArgsMsg) NotRunning() SnapsCond

func (*ArgsMsg) Started

func (args *ArgsMsg) Started() SnapsCond

func (*ArgsMsg) String

func (args *ArgsMsg) String() string

type Base

type Base struct {
	// contains filtered or unexported fields
}

func (*Base) Abort

func (xctn *Base) Abort(err error) bool

func (*Base) AbortErr

func (xctn *Base) AbortErr() error

func (*Base) AbortedAfter

func (xctn *Base) AbortedAfter(d time.Duration) (err error)

func (*Base) AddErr

func (xctn *Base) AddErr(err error, logExtra ...int)

func (*Base) AddNotif

func (xctn *Base) AddNotif(n core.Notif)

func (*Base) BcastCtrl

func (xctn *Base) BcastCtrl(smap *meta.Smap, wid, opcode string, err error)

asynchronous broadcast with bounded launch parallelism (note: NOT waiting for completion)

func (*Base) Bck

func (xctn *Base) Bck() *meta.Bck

func (*Base) Bytes

func (xctn *Base) Bytes() int64

func (*Base) ChanAbort

func (xctn *Base) ChanAbort() <-chan error

func (*Base) Cname

func (xctn *Base) Cname() string

func (*Base) EndTime

func (xctn *Base) EndTime() time.Time

func (*Base) Err

func (xctn *Base) Err() (err error)

func (*Base) ErrCnt

func (xctn *Base) ErrCnt() int

func (*Base) Finish

func (xctn *Base) Finish()

func (*Base) FromTo

func (*Base) FromTo() (*meta.Bck, *meta.Bck)

func (*Base) ID

func (xctn *Base) ID() string

func (*Base) InBytes

func (xctn *Base) InBytes() int64

func (*Base) InObjs

func (xctn *Base) InObjs() int64

base stats: receive

func (*Base) InObjsAdd

func (xctn *Base) InObjsAdd(cnt int, size int64)

func (*Base) InitBase

func (xctn *Base) InitBase(id, kind string, bck *meta.Bck)

func (*Base) IsAborted

func (xctn *Base) IsAborted() bool

func (*Base) IsDone

func (xctn *Base) IsDone() bool

return true if 'stopping' OR 'finished'

func (*Base) IsIdle

func (xctn *Base) IsIdle() bool

NOTE on existing legacy: * default idle == not-running * Demand-based xactions override as needed * Python SDK's Job.wait_for_idle() and, potentially, other callers currently expect this behavior * this MAY change in the future

func (*Base) IsRunning

func (xctn *Base) IsRunning() (yes bool)

func (*Base) JoinErr

func (xctn *Base) JoinErr() (int, error)

func (*Base) Kind

func (xctn *Base) Kind() string

func (*Base) LomAdd

func (xctn *Base) LomAdd(lom *core.LOM)

oft. used

func (*Base) Name

func (xctn *Base) Name() string

func (*Base) NewErrRecvAbortWI

func (xctn *Base) NewErrRecvAbortWI(tid, wid, errCause string) error

func (*Base) NewErrRecvAbortXact

func (xctn *Base) NewErrRecvAbortXact(tid, errCause string) error

func (*Base) NewSnap

func (xctn *Base) NewSnap(self core.Xact) (snap *core.Snap)

func (*Base) Objs

func (xctn *Base) Objs() int64

base stats: locally processed

func (*Base) ObjsAdd

func (xctn *Base) ObjsAdd(cnt int, size int64)

func (*Base) OutBytes

func (xctn *Base) OutBytes() int64

func (*Base) OutObjs

func (xctn *Base) OutObjs() int64

base stats: transmit

func (*Base) OutObjsAdd

func (xctn *Base) OutObjsAdd(cnt int, size int64)

func (*Base) Quiesce

func (xctn *Base) Quiesce(d time.Duration, cb core.QuiCB) core.QuiRes

count all the way to duration; reset and adjust every time activity is detected

func (*Base) SendCtrl

func (xctn *Base) SendCtrl(tsi *meta.Snode, wid, opcode string, body []byte) error

func (*Base) SetStopping

func (xctn *Base) SetStopping() bool

mark the xaction as stopping (and finishing soon); callers that perform terminal cleanup can use the return value to decide whether they own that cleanup path

func (*Base) StartTime

func (xctn *Base) StartTime() time.Time

func (*Base) String

func (xctn *Base) String() string

func (*Base) ToStats

func (xctn *Base) ToStats(stats *core.Stats)

type BckJog

type BckJog struct {
	Config *cmn.Config

	Base
	// contains filtered or unexported fields
}

func (*BckJog) Init

func (r *BckJog) Init(id, kind string, bck *meta.Bck, opts *mpather.JgroupOpts, config *cmn.Config)

func (*BckJog) NumJoggers

func (r *BckJog) NumJoggers() int

func (*BckJog) NumVisits

func (r *BckJog) NumVisits() int64

func (*BckJog) Run

func (r *BckJog) Run()

func (*BckJog) Wait

func (r *BckJog) Wait() error

type BckJogRunner

type BckJogRunner struct {
	BckJog
	// contains filtered or unexported fields
}

BckJogRunner extends BckJog with a managed worker pool. The caller provides a single CbObj callback; BckJogRunner owns dispatch, channel management, slab allocation, and worker lifecycle.

func (*BckJogRunner) Init

func (r *BckJogRunner) Init(id, kind string, bck *meta.Bck, opts BckJogRunnerOpts, config *cmn.Config) error

Init initializes BckJogRunner from opts. If opts.NumWorkers != NwpNone, it auto-tunes worker count (media + load) and sets up the internal worker pool. Returns an error only if the system is under such extreme pressure that starting workers is unsafe.

func (*BckJogRunner) NumWorkers

func (r *BckJogRunner) NumWorkers() int

NumWorkers returns the configured worker count (0 if no pool).

func (*BckJogRunner) Run

func (r *BckJogRunner) Run()

Run launches workers (if any) then starts the mountpath joggers.

func (*BckJogRunner) Wait

func (r *BckJogRunner) Wait() error

Wait waits for joggers to finish then drains the worker pool.

func (*BckJogRunner) WorkChanFull

func (r *BckJogRunner) WorkChanFull() int64

WorkChanFull returns the accumulated count of work-channel-full events.

type BckJogRunnerOpts

type BckJogRunnerOpts struct {
	// CbObj is called for each visited object.
	CbObj func(*core.LOM, []byte) error

	// WalkBck, when non-nil, overrides the traversal bucket.
	// Use when the xaction's logical bucket differs from the walk source
	// (e.g. copy-bucket registers the destination but walks the source).
	WalkBck *meta.Bck

	// Prefix filters visited objects by name prefix.
	Prefix string

	// RW must be true when the xaction modifies the local filesystem.
	RW bool

	// NumWorkers controls worker pool size:
	//   NwpNone (-1): no pool, objects processed inline
	//   NwpDflt (0):  auto-tune based on media type and current load
	//   > 0:          treat as a ceiling; still throttled under load
	NumWorkers int

	// Burst sets the work-channel lower bound (minimum capacity).
	// Zero means use cmn.XactBurstDflt. Callers with their own burst config
	// (e.g. TCB uses config.TCB.Burst) should pass it explicitly.
	Burst int
}

BckJogRunnerOpts configures BckJogRunner initialization.

type Demand

type Demand interface {
	core.Xact
	IdleTimer() <-chan struct{}
	IncPending()
	DecPending()
	SubPending(n int)
}

xaction that self-terminates after staying idle for a while with an added capability to renew itself and ref-count its pending work

type DemandBase

type DemandBase struct {
	Base
	// contains filtered or unexported fields
}

func (*DemandBase) Abort

func (r *DemandBase) Abort(err error) (ok bool)

Only abort the Base part, and note: for demand xactions, Stop() is a cleanup-sequence sensitive barrier: it unregisters the idle HK callback and can allow same-UUID renewal to register the same HK name.

func (*DemandBase) DecPending

func (r *DemandBase) DecPending()

func (*DemandBase) IdleTimer

func (r *DemandBase) IdleTimer() <-chan struct{}

func (*DemandBase) IncPending

func (r *DemandBase) IncPending()

func (*DemandBase) Init

func (r *DemandBase) Init(uuid, kind string, bck *meta.Bck, idleDur time.Duration, hkcb ...hk.HKCB)

func (*DemandBase) IsIdle

func (r *DemandBase) IsIdle() bool

NOTE: override `Base.IsIdle`

func (*DemandBase) Pending

func (r *DemandBase) Pending() (cnt int64)

func (*DemandBase) Reset

func (r *DemandBase) Reset(idleTime time.Duration)

(e.g. usage: listed last page)

func (*DemandBase) Stop

func (r *DemandBase) Stop()

func (*DemandBase) SubPending

func (r *DemandBase) SubPending(n int)

type Descriptor

type Descriptor struct {
	DisplayName string          // as implied
	Access      apc.AccessAttrs // default access permissions; ais/proxy does most of the checking wo/ relying on these defaults
	Scope       int             // ScopeG (global), etc. - the enum above
	Startable   bool            // true if user can start this xaction (e.g., via `api.StartXaction`)
	Metasync    bool            // true if this xaction changes (and metasyncs) cluster metadata
	RefreshCap  bool            // refresh capacity stats upon completion

	// see xreg for "limited coexistence"
	Rebalance      bool // moves data between nodes
	Resilver       bool // moves data between mountpaths
	ConflictRebRes bool // starting this job would conflict with rebalance or resilver that's currently in progress
	AbortByReb     bool // gets aborted upon rebalance (coincides with ConflictRebRes with very few exceptions)

	// xaction has an intermediate `idle` state whereby it "idles" between requests
	// (see related: xact/demand.go)
	Idles bool

	// xaction returns extended xaction-specific stats
	// (see related: `Snap.Ext` in core/xaction.go)
	ExtendedStats bool

	// suppress verbose per-state log records and keep only hk.OldAgeXshort (1m)
	// in registry history
	QuietBrief bool

	// IC reporting mode; see ICMode comment above
	ICMode ICMode
}

func GetDescriptor

func GetDescriptor(kindOrName string) (string, Descriptor, error)

type ICMode

type ICMode uint8

ICMode declares whether and how an xaction kind reports status to IC. When non-zero, targets notify IC members, and the status/wait API may query an IC proxy instead of polling all targets. When zero, the generic IC status path is not available; callers must use the snaps-based API or an action-specific status path.

The flags are independent and may be combined: * ICUponTerm - target notifies IC on terminal state (finished/aborted); * ICUponProgress - target notifies IC periodically with progress

const (
	ICUponTerm     ICMode = 1 << iota // -> core.UponTerm
	ICUponProgress                    // -> core.UponProgress
)

const ICNone ICMode = 0

type Marked

type Marked struct {
	Xact        core.Xact
	Interrupted bool // (rebalance | resilver) interrupted
	Restarted   bool // node restarted
}

type MultiSnap

type MultiSnap map[string][]*core.Snap // by target ID (tid)

`api.QueryXactionSnaps` control structure

func (MultiSnap) AggregateState

func (xs MultiSnap) AggregateState(xid string) (aborted, running, notstarted bool)

return: `aborted` => any selected xaction aborted on any target `running` => any selected xaction currently running on any target `notstarted` => selected xaction not yet visible / not started on any target selection: - xid != "": the specified xaction UUID (all targets) - xid == "": all UUIDs present in this MultiSnap (all targets)

func (MultiSnap) ByteCounts

func (xs MultiSnap) ByteCounts(xid string) (locBytes, outBytes, inBytes int64)

func (MultiSnap) GetUUIDs

func (xs MultiSnap) GetUUIDs() []string

func (MultiSnap) ObjCounts

func (xs MultiSnap) ObjCounts(xid string) (locObjs, outObjs, inObjs int64)

func (MultiSnap) RunningTarget

func (xs MultiSnap) RunningTarget(xid string) (string, *core.Snap, error)

func (MultiSnap) ToJSON

func (xs MultiSnap) ToJSON(tid string, indent bool) ([]byte, error)

func (MultiSnap) TotalRunningTime

func (xs MultiSnap) TotalRunningTime(xid string) (time.Duration, error)

type NotifXact

type NotifXact struct {
	Xact core.Xact
	nl.Base
}

func (*NotifXact) ToNotifMsg

func (nx *NotifXact) ToNotifMsg(aborted bool) core.NotifMsg

type NotifXactListener

type NotifXactListener struct {
	nl.ListenerBase
}

func NewXactNL

func NewXactNL(uuid, kind string, smap *meta.Smap, srcs meta.NodeMap, bck ...*cmn.Bck) *NotifXactListener

func (*NotifXactListener) QueryArgs

func (nxb *NotifXactListener) QueryArgs() cmn.HreqArgs

func (*NotifXactListener) UnmarshalStats

func (*NotifXactListener) UnmarshalStats(rawMsg []byte) (stats any, finished, aborted bool, err error)

func (*NotifXactListener) WithCause

func (nxb *NotifXactListener) WithCause(cause string) *NotifXactListener

type QueryMsg

type QueryMsg struct {
	OnlyRunning *bool     `json:"show_active"`
	Bck         cmn.Bck   `json:"bck"`
	ID          string    `json:"id"`
	Kind        string    `json:"kind"`
	DaemonID    string    `json:"node,omitempty"`
	Buckets     []cmn.Bck `json:"buckets,omitempty"`
}

simplified JSON-tagged version of the ArgsMsg (internal use)

func (*QueryMsg) String

func (msg *QueryMsg) String() (s string)

type SnapsCond

type SnapsCond func(MultiSnap) (done, reset bool, err error)

SnapsCond is the condition function signature for snaps-based polling. Returns:

• done: whether to stop waiting
• reset: whether to reset polling sleep back to MinPollTime
• err: error to propagate (stops polling immediately)