Method assembleStatusFromMetadata

usecases/export/scheduler.go:529–599

assembleStatusFromMetadata reads per-node status files from the configured backup backend and assembles overall status using the metadata's NodeAssignments. The returned allTerminal flag is true when every node has written a status file with a terminal status (Success or Failed), distinguishing genu

(
	ctx context.Context,
	backend modulecapabilities.BackupBackend,
	id, bucket, path string,
	meta *ExportMetadata,
)

// written a status file with a terminal status (Success or Failed),
// distinguishing genuine completions from liveness-inferred failures.
func (s *Scheduler) assembleStatusFromMetadata(
	ctx context.Context,
	backend modulecapabilities.BackupBackend,
	id, bucket, path string,
	meta *ExportMetadata,
) (_ *models.ExportStatusResponse, allTerminal bool, _ error) {
	homePath := backend.HomeDir(id, bucket, path)

	// Read per-node statuses and apply liveness overrides for non-terminal nodes.
	// The status file is read up to twice per node. The export goroutine writes
	// the terminal status and then clears its activeExport flag. An IsRunning
	// check that lands between those two events sees "not running" while the
	// status file still shows Transferring/Started. A single re-read is
	// sufficient because stopWriter blocks until the write completes before
	// clearAndRelease clears activeExport, so the file is guaranteed to be on
	// disk when IsRunning returns false.
	const maxStatusReads = 2
	nodeStatuses := make(map[string]*NodeStatus, len(meta.NodeAssignments))
	for nodeName := range meta.NodeAssignments {
		var nodeStatus *NodeStatus
		for attempt := range maxStatusReads {
			ns, err := readNodeStatus(ctx, backend, id, bucket, path, nodeName)
			if err != nil {
				if !errors.As(err, &backup.ErrNotFound{}) {
					return nil, false, fmt.Errorf("get status for node %s: %w", nodeName, err)
				}
				ns = &NodeStatus{
					NodeName:      nodeName,
					Status:        export.Transferring,
					ShardProgress: make(map[string]map[string]*ShardProgress),
				}
			}

			// For non-terminal nodes, verify liveness and override to Failed if unreachable.
			switch ns.Status {
			case export.Success, export.Failed:
				// Terminal — no liveness check needed.
				nodeStatus = ns
			default:
				host, alive := s.nodeResolver.NodeHostname(nodeName)
				if !alive {
					ns.Status = export.Failed
					if ns.Error == "" {
						ns.Error = fmt.Sprintf("node %s is no longer part of the cluster", nodeName)
					}
					nodeStatus = ns
				} else {
					running, runErr := s.client.IsRunning(ctx, host, id)
					if runErr != nil || !running {
						if attempt < maxStatusReads-1 {
							// Re-read the status file — the goroutine may
							// have written the terminal status by now.
							continue
						}
						ns.Status = export.Failed
						if ns.Error == "" {
							ns.Error = fmt.Sprintf("node %s is no longer running export %s", nodeName, id)
						}