myxce/src/frontend/apiserver/backend/handler.go

package backend

import (
	"apiserver/models"
	"apiserver/store"
	logger "apiserver/utils"
	"sync"

	"go.uber.org/zap"
)

// DefaultHandler implements NotificationHandler by writing to stores.
type DefaultHandler struct{}

// defaultHostConfigName is the cluster-wide config sentinel in hosts.yaml
// (never a real worker) whose fields worker.c's apply_join_state()/
// find_matched_hostconfig() already falls back to on the C side for any
// worker with no host-specific hosts.yaml entry of its own.
const defaultHostConfigName = "default"

// bug-1673: a dynamically-joined worker with no hosts.yaml entry under its
// own literal hostname never receives its own OnHostConfigUpdated call, so
// MaxSlot/HostModel/HostType/DisptWindow stayed at Go's zero-value forever
// -- decoupled from the C-side scheduling path, which already resolves the
// real value via the "default" wildcard entry. These package-level vars
// cache that entry's fields plus which hosts have their OWN explicit
// (non-default) config, so OnHostUpdated can apply the same fallback for
// display purposes that worker.c already applies for real dispatch.
var (
	hostConfigMu            sync.Mutex
	defaultHostConfig       *models.HostInfo
	hostsWithExplicitConfig = map[string]bool{}
)

// resetHostConfigCache clears the cached "default" pseudo-host config and
// explicit-config tracking. Test-only: called from this package's _test.go
// store-reset helpers, so bug-1673 fallback state doesn't leak between
// tests the way store.InitStores() already prevents for the stores
// themselves.
func resetHostConfigCache() {
	hostConfigMu.Lock()
	defaultHostConfig = nil
	hostsWithExplicitConfig = map[string]bool{}
	hostConfigMu.Unlock()
}

// applyDefaultHostConfigFallback copies the cached "default" pseudo-host's
// config-sourced fields onto host, if one has been observed yet. No-op
// otherwise (nothing to fall back to).
//
// A negative MaxSlot is HOSTCONFIG_MXJ_AUTO_DETECT (bug-1556): "no explicit
// cap, use this host's real CPU count" -- the same resolution worker.c's
// apply_join_state()/find_matched_hostconfig() already performs for actual
// dispatch ("if (nslots<0) nslots = maxcpus"). Resolving it here too means
// this display fallback shows the host's real usable capacity (matching
// what dispatch will actually honor) instead of a bare "-1" that would
// leave the reporter's original "is ncpus equal slots?" question (gitea
// #40) just as unanswered as the "0" it replaces.
func applyDefaultHostConfigFallback(host *models.HostInfo) {
	hostConfigMu.Lock()
	def := defaultHostConfig
	hostConfigMu.Unlock()
	if def == nil {
		return
	}
	host.HostModel = def.HostModel
	host.HostType = def.HostType
	host.MaxSlot = def.MaxSlot
	if host.MaxSlot < 0 && host.NumCPUs > 0 {
		host.MaxSlot = host.NumCPUs
	}
	host.DisptWindow = def.DisptWindow
}

func (h *DefaultHandler) OnJobCreated(job *models.JobInfo) {
	store.Jobs.Put(uint64(job.Unique_JobID), *job)
	DefaultBroker.Publish(Event{Type: EventJobCreated, JobID: uint64(job.Unique_JobID), Status: models.JobStateStr(job.JobStatus), Job: job})
}

func (h *DefaultHandler) OnJobUpdated(job *models.JobInfo) {
	store.Jobs.Put(uint64(job.Unique_JobID), *job)
	DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: uint64(job.Unique_JobID), Status: models.JobStateStr(job.JobStatus), Job: job})
}

func (h *DefaultHandler) OnJobDeleted(jobid uint64) {
	store.Jobs.Delete(jobid)
	DefaultBroker.Publish(Event{Type: EventJobDeleted, JobID: jobid})
}

// bug-1731: previously permanently starved of data -- apply_committed_entry()
// (persist.c) never fired the controller-side HOOK_DETAIL_JOB_EXIT this
// method's C caller (apiserver_job_exit_fun -> NotifyData_data_jobFinishInfo)
// depends on, so OnJobFinished was live, wired, and correct but never once
// invoked in production. Logs at Info (mirroring this file's existing
// "job array push received and expanded" convention) so a finished job's
// terminal-state push to the apiserver is directly observable in
// apiserver.log, not just inferable from store state.
func (h *DefaultHandler) OnJobFinished(jobid uint64, exitCode int) {
	existing, ok := store.Jobs.Get(jobid)
	if !ok {
		logger.Info("OnJobFinished: jobid not found in store, ignoring",
			zap.Uint64("jobid", jobid), zap.Int("exit_code", exitCode))
		return
	}
	if exitCode == 0 {
		existing.JobStatus = models.JOBSTAT["DONE"]
	} else {
		existing.JobStatus = models.JOBSTAT["EXIT"]
	}
	store.Jobs.Put(jobid, existing)
	DefaultBroker.Publish(Event{Type: EventJobFinished, JobID: jobid, ExitCode: exitCode})

	logger.Info("OnJobFinished: job status updated",
		zap.Uint64("jobid", jobid), zap.Int("exit_code", exitCode),
		zap.Int("job_status", existing.JobStatus))
}

// OnJobPidInfo applies a controller-pushed job start report
// (NotifyData_data_jobPidInfo, fired from HOOK_DETAIL_JOB_START after
// WAL_JOB_START commits) onto the already-stored job record: the worker's
// pids/pgids become PIDInfo, and an interactive job's exdata {"url": ...}
// becomes InteractiveJobURL — the field xcecli's waitForInteractiveURL
// polls. Without this handler the push was silently dropped by
// handleNotifyData and the URL stayed invisible until the next periodic
// rusage-driven job UPDATE re-delivered the full row (up to 10s later,
// job_rusage_collect_interval).
//
// Mirrors OnJobFinished's get-mutate-put-publish shape: the full job row
// (submit spec, state, timestamps) always arrives ahead of or with the
// start report — SUBMIT via OnJobCreated/join push, RUN state itself via
// the dispatch-time update — so a jobid not in the store is logged and
// dropped rather than inserted half-empty.
func (h *DefaultHandler) OnJobPidInfo(jobid uint64, pidInfo *rawPidInfo) {
	existing, ok := store.Jobs.Get(jobid)
	if !ok {
		logger.Info("OnJobPidInfo: jobid not found in store, ignoring",
			zap.Uint64("jobid", jobid))
		return
	}
	if pidInfo != nil {
		if jb, ok := pidInfoJSONB(pidInfo.pids, pidInfo.pgids); ok {
			existing.PIDInfo = jb
		}
		existing.InteractiveJobURL = interactiveURLFromExdata(pidInfo.exdata)
	}
	store.Jobs.Put(jobid, existing)
	DefaultBroker.Publish(Event{Type: EventJobUpdated, JobID: jobid,
		Status: models.JobStateStr(existing.JobStatus), Job: &existing})

	logger.Info("OnJobPidInfo: start report applied",
		zap.Uint64("jobid", jobid),
		zap.Int("pids", len(pidInfo.pids)),
		zap.String("interactive_job_url", existing.InteractiveJobURL))
}

func (h *DefaultHandler) OnQueueUpdated(queue *models.QueueInfo) {
	store.Queues.Put(queue.QueueName, *queue)
}

// OnHostUpdated merges worker-sourced fields into an existing host entry.
// Config-sourced fields (HostModel, HostType, MaxSlot, DisptWindow) are
// preserved from the prior entry if one exists. If this is the host's
// first-ever registration (no prior entry), fall back to the cached
// "default" pseudo-host's config (bug-1673) rather than leaving those
// fields at Go's zero-value -- mirrors the C-side wildcard fallback.
func (h *DefaultHandler) OnHostUpdated(incoming *models.HostInfo) {
	merged := *incoming
	if existing, ok := store.Hosts.Get(merged.HostName); ok {
		merged.HostModel = existing.HostModel
		merged.HostType = existing.HostType
		merged.MaxSlot = existing.MaxSlot
		merged.DisptWindow = existing.DisptWindow
	} else {
		applyDefaultHostConfigFallback(&merged)
	}
	store.Hosts.Put(merged.HostName, merged)
}

// OnHostConfigUpdated applies config-sourced fields onto an existing worker-registered host.
// Hosts not yet registered by a worker are skipped — config entries like "default" (which sets
// cluster-wide defaults) or hosts whose daemons haven't started must not appear in the host list.
//
// The "default" pseudo-host entry itself is cached (bug-1673) rather than
// simply discarded, then retroactively backfilled onto any already-
// registered host with no explicit config of its own -- "default" can
// arrive either before or after a given worker's own OnHostUpdated call.
func (h *DefaultHandler) OnHostConfigUpdated(incoming *models.HostInfo) {
	if incoming.HostName == defaultHostConfigName {
		hostConfigMu.Lock()
		cp := *incoming
		defaultHostConfig = &cp
		hostConfigMu.Unlock()

		store.Hosts.Range(func(name string, host models.HostInfo) bool {
			hostConfigMu.Lock()
			explicit := hostsWithExplicitConfig[name]
			hostConfigMu.Unlock()
			if !explicit {
				applyDefaultHostConfigFallback(&host)
				store.Hosts.Put(name, host)
			}
			return true
		})
		return
	}

	hostConfigMu.Lock()
	hostsWithExplicitConfig[incoming.HostName] = true
	hostConfigMu.Unlock()

	existing, ok := store.Hosts.Get(incoming.HostName)
	if !ok {
		return
	}
	existing.HostModel = incoming.HostModel
	existing.HostType = incoming.HostType
	existing.MaxSlot = incoming.MaxSlot
	existing.DisptWindow = incoming.DisptWindow
	store.Hosts.Put(existing.HostName, existing)
}

// OnClusterUpdated applies config-sourced fields onto the existing cluster entry.
// All other fields (built-in/worker-sourced: Version, Status, ControllerName,
// NumServers, Resources, HostTypes, HostModels, ClusterAdmins) are preserved
// from the prior entry.
func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) {
	existing, _ := store.Clusters.Get(incoming.ClusterName)
	existing.ClusterName = incoming.ClusterName
	if incoming.HeartbeatInterval > 0 {
		existing.HeartbeatInterval = incoming.HeartbeatInterval
	}
	existing.Vcpus = incoming.Vcpus
	existing.InactiveHeartbeatCount = incoming.InactiveHeartbeatCount
	store.Clusters.Put(existing.ClusterName, existing)
	SetHeartbeatInterval(incoming.HeartbeatInterval)
}

// OnUserGroupsUpdated replaces the stored user groups as a whole.
func (h *DefaultHandler) OnUserGroupsUpdated(groups []models.UserGroupInfo) {
	store.SetUserGroups(groups)
}

func (h *DefaultHandler) OnResourceUpdated(res *models.ResourceDefinitionInfo) {
	store.Resources.Put(res.ResourceName, *res)
}

func (h *DefaultHandler) OnResourceLocationUpdated(res *models.ResourceDistributionInfo) {
	key := res.ResourceName + ":" + res.Location
	store.ResourceLocations.Put(key, *res)

	// resource-value-verification task 4.2 (Requirements 5.1, 5.2, 5.4): trace
	// the stored value at its point of application, mirroring the scheduler's
	// sibling trace log added in task 4.1 (core_events.c). One typed "value"
	// field carrying whichever of StringValue/NumericValue/BooleanValue is
	// live for res.ValueType, so a downstream grep sees exactly what was
	// stored, not every possibly-zero field on the struct.
	//
	// DO NOT change the "resource-location stored" message string, or the
	// resource/host/value field names below, without also updating
	// query_apiserver_resvals() in
	// .claude/skills/xce-test-resources/run-resource-verify.sh (design.md's
	// own Revalidation Triggers list this exact pair) -- that script greps
	// this exact message and parses zap's console-encoder trailing JSON
	// block for these field names; an unrelated logging cleanup here
	// silently turns every future verification run into a false
	// UNREACHABLE/timeout, indistinguishable from a real regression.
	var valueField zap.Field
	switch res.ValueType {
	case int(models.STRING):
		valueField = zap.String("value", res.StringValue)
	case int(models.BOOLEAN):
		valueField = zap.Bool("value", res.BooleanValue)
	default:
		valueField = zap.Float64("value", res.NumericValue)
	}
	logger.Debug("resource-location stored",
		zap.String("resource", res.ResourceName),
		zap.String("host", res.Location),
		valueField)
}

// dispatchJobArray delivers a locally-expanded compact array push to the
// handler, one call per genuinely-new element (job-array-single-wal-record
// task 5.4, Req 6.2, 6.4, 6.5, 8.4). CGO-free (called from handleDataset's
// XceData_data_jobArray case in clib.go; the elements themselves are already
// decoded by cJobArrayToModels/rawJobArrayToModels before this is called),
// mirroring dispatchConfigPush's split for testability without CGO.
//
// Dedup reuses store.Jobs itself — the SAME jobid-keyed store the ordinary
// single-job push path already writes through (handleDataset's
// XceData_data_jobs case calls OnJobUpdated per job with no separate dedup
// step, relying on the map being keyed by jobid) — checked here BEFORE
// calling the handler so a redelivered push (e.g. after an apiserver
// reconnect) can never clobber a jobid whose state has since moved on.
// This mirrors the scheduler's sched_on_job_array() (handler.c) skip-if-
// exists behavior exactly, not a new dedup mechanism.
//
// Logs exactly one confirmation line (base jobid + array size + new-element
// count) when at least one element is genuinely new; a fully-duplicate
// redelivery (every index already exists) produces zero log lines, matching
// task 5.3's resolved semantics for the equivalent scheduler-side push.
// Returns the count of newly-dispatched elements (test-observable).
func dispatchJobArray(handler NotificationHandler, elements []*models.JobInfo, baseJobID, arraySize uint64) int {
	if handler == nil {
		return 0
	}

	nNew := 0
	for _, job := range elements {
		jobid := uint64(job.Unique_JobID)
		// A provisional record (bug-1894) is not a duplicate: replace it.
		if existing, ok := store.Jobs.Get(jobid); ok && !existing.Provisional {
			continue
		}
		handler.OnJobUpdated(job)
		nNew++
	}

	if nNew > 0 {
		logger.Info("job array push received and expanded",
			zap.String("array_jobid", FormatJobID(int64(ExtractArrayBaseID(baseJobID)))),
			zap.Uint64("array_size", arraySize),
			zap.Int("new_records", nNew))
	}

	return nNew
}

// dispatchConfigPush delivers converted config-push data to the handler (CGO-free, for testability).
func dispatchConfigPush(
	handler NotificationHandler,
	queues []models.QueueInfo,
	resources []models.ResourceDefinitionInfo,
	hostConfigs []models.HostInfo,
	cluster *models.ClusterInfo,
) {
	for _, q := range queues {
		handler.OnQueueUpdated(&q)
	}
	for _, r := range resources {
		handler.OnResourceUpdated(&r)
	}
	for _, hc := range hostConfigs {
		handler.OnHostConfigUpdated(&hc)
	}
	if cluster != nil {
		handler.OnClusterUpdated(cluster)
	}
}