diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 39975ac..5701689 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -140,6 +140,9 @@ class LoadedNode: #: so only an acknowledgement clears it, not a good run and not a rebuild. last_error: str = "" last_error_ts: float | None = None + #: Published, and with no body of its own — so what runs is the new-node + #: template, which returns nothing at all. + missing_source: bool = False @dataclass @@ -972,6 +975,12 @@ class FlowController: else: owner, local = flow, node_def.id code = self.store.read_node_source(flow, node_def.id, draft=draft) + # The store answers the new-node template when nothing was + # written, which is what an editor should open with and not + # something to run. A draft legitimately has none yet. + entry.missing_source = not draft and not self.store.has_node_source( + flow, node_def.id + ) # What a node produces before it returns comes back as frames; # this puts them through the node's own ports. @@ -1679,6 +1688,21 @@ def _collect_issues( and entry.node.mode == HttpNode.Mode.TRIGGER and not entry.node.secret ] + # A node whose body was never stored runs the new-node template, which + # returns nothing — quietly, and looking healthy the whole time. + issues += [ + ValidationIssue( + code="missing_source", + message=( + f"Node '{entry.id.rpartition('.')[2]}' has no stored code. It " + "runs as an empty node and publishes nothing." + ), + flow=entry.flow, + node=entry.id, + ) + for entry in loaded.values() + if entry.missing_source + ] return issues diff --git a/backend/fluksio/flow/pipeline.py b/backend/fluksio/flow/pipeline.py index 01ba8a4..7bace47 100644 --- a/backend/fluksio/flow/pipeline.py +++ b/backend/fluksio/flow/pipeline.py @@ -58,6 +58,7 @@ class ValidationIssue(BaseModel): "node_error", "unauthenticated_hook", "self_loop_needs_initial", + "missing_source", ] message: str flow: str = "" diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 18e534e..1e21a75 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -48,12 +48,13 @@ from sqlmodel import Session, col, select from fluksio.core.db import engine as db_engine from fluksio.flow.artifacts import ArtifactStore, is_reference, valid_digest -from fluksio.flow.controller import FlowController, RunContext +from fluksio.flow.controller import NODE_TYPES, FlowController, RunContext from fluksio.flow.messages import qualify from fluksio.flow.pipeline import CacheHit, NodeOutcome, Pipeline from fluksio.flow.queue import WorkItem, WorkQueue -from fluksio.flow.schemas import FlowDef +from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.state import MemoryState, StateBackend +from fluksio.flow.store import FlowStore from fluksio.models import Run, RunArtifact, RunMetric, RunNode logger = logging.getLogger(__name__) @@ -106,7 +107,9 @@ def digest_of(params: dict[str, Any], seed: int | None) -> str: return hashlib.sha256(canonical.encode()).hexdigest() -def batch_issues(flow: FlowDef) -> list[str]: +def batch_issues( + flow: FlowDef, store: FlowStore | None = None, draft: bool = False +) -> list[str]: """Why this flow cannot be run as a batch, if it cannot. One thing genuinely breaks: a port with a discretization interval holds @@ -118,6 +121,12 @@ def batch_issues(flow: FlowDef) -> list[str]: A delay node is fine; without a queue to defer into it simply sleeps, which in a run is what was asked for. + + The other is a node with no body. The store answers a new node's template + when nothing was ever written for one, so such a node runs — and returns + ``{}`` every time, without a word. `fluksio sync` writes every body before + it publishes, so this is unreachable from there; it is the other end that + is open. """ issues: list[str] = [] for node in flow.nodes: @@ -129,9 +138,28 @@ def batch_issues(flow: FlowDef) -> list[str]: "the value would be dropped. Remove the interval, or mark " "the port as streaming if it is a curve being thinned out." ) + if store is not None and _has_no_body(store, flow.name, node, draft): + issues.append( + f"Node '{node.id}' has no stored code. It would run as an " + "empty node and publish nothing, so the run is refused." + ) return issues +def _has_no_body( + store: FlowStore, flow: str, node: NodeDef, draft: bool = False +) -> bool: + """A node that should carry its own source, and does not. + + A shared node runs the library's copy, so it is not this: a missing library + fails the node loudly on its own. + """ + node_type = NODE_TYPES.get(node.type) + if node_type is None or not node_type.has_source or node.source_ref: + return False + return not store.has_node_source(flow, node.id, draft=draft) + + def required_labels(flow: FlowDef) -> list[str]: """Worker labels this flow cannot run without. @@ -644,7 +672,7 @@ class RunService: if existing is not None: return existing flow = self.controller.store.read_flow(flow_name, draft=draft) - issues = batch_issues(flow) + issues = batch_issues(flow, self.controller.store, draft=draft) if issues: raise RunRejected(" ".join(issues)) params = params or {} diff --git a/backend/tests/flow/test_drafts.py b/backend/tests/flow/test_drafts.py index 2dbafde..b3d085f 100644 --- a/backend/tests/flow/test_drafts.py +++ b/backend/tests/flow/test_drafts.py @@ -1,9 +1,11 @@ """Editing writes drafts; only publishing changes what the engine reads.""" +import asyncio from pathlib import Path import pytest +from fluksio.flow.controller import FlowController from fluksio.flow.messages import MessageSpec from fluksio.flow.schemas import FlowDef, NodeDef from fluksio.flow.store import FlowStore, StaleVersion @@ -106,6 +108,31 @@ def test_an_edited_source_alone_counts_as_a_draft(store: FlowStore): assert store.read_node_source("heating", "sensor") == EDITED +def test_a_published_node_with_no_body_is_reported(store: FlowStore): + """It would run the new-node template, which returns nothing and says so.""" + store.write_flow(a_flow()) + controller = FlowController(store) + + asyncio.run(controller.reload()) + (issue,) = [i for i in controller.issues if i.code == "missing_source"] + assert issue.node == "heating.sensor" + assert not issue.advisory + # It still loads: one node with no body does not take the flow down. + assert controller.get_node("heating.sensor") is not None + + store.write_node_source("heating", "sensor", SOURCE) + asyncio.run(controller.reload()) + assert [i for i in controller.issues if i.code == "missing_source"] == [] + + +def test_a_node_being_written_in_the_editor_is_not_reported(store: FlowStore): + """A draft legitimately has no published body yet — that is what a draft is.""" + store.write_draft(a_flow(), 0) + controller = FlowController(store) + + assert controller.preview("heating").issues == [] + + def test_resaving_the_published_source_creates_no_draft(store: FlowStore): store.write_flow(a_flow()) store.write_node_source("heating", "sensor", SOURCE) diff --git a/backend/tests/flow/test_runs.py b/backend/tests/flow/test_runs.py index 9b06221..59aeacb 100644 --- a/backend/tests/flow/test_runs.py +++ b/backend/tests/flow/test_runs.py @@ -317,6 +317,25 @@ def test_a_rate_limited_port_cannot_be_run_as_a_batch(): assert not batch_issues(double_flow()) +def test_a_node_with_no_stored_code_cannot_be_run(): + """The store answers a template for one, and a template publishes nothing.""" + + class Store: + def __init__(self, *, written: bool): + self.written = written + + def has_node_source(self, flow, node_id, draft=False): + return self.written + + flow = double_flow() + (issue,) = batch_issues(flow, Store(written=False)) + assert "no stored code" in issue + assert not batch_issues(flow, Store(written=True)) + # A shared node runs the library's copy, so it is not this node's to have. + flow.nodes[0].source_ref = "shared" + assert not batch_issues(flow, Store(written=False)) + + def test_a_streaming_port_may_thin_itself_out(): flow = double_flow() flow.nodes[0].provides = [spec("loss", interval=0.5, stream=True)] diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index c8170be..09dbe2c 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -1662,6 +1662,17 @@ export const NodeDef_InputSchema = { title: 'Cache', description: 'Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own.', default: true + }, + resources: { + anyOf: [ + { + '$ref': '#/components/schemas/Resources' + }, + { + type: 'null' + } + ], + description: "What one execution of this node holds while it runs. Absent — the default — means it is not accounted for and shares the engine's workers, which is right for everything that is not compute-heavy." } }, type: 'object', @@ -1757,6 +1768,17 @@ export const NodeDef_OutputSchema = { title: 'Cache', description: 'Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own.', default: true + }, + resources: { + anyOf: [ + { + '$ref': '#/components/schemas/Resources' + }, + { + type: 'null' + } + ], + description: "What one execution of this node holds while it runs. Absent — the default — means it is not accounted for and shares the engine's workers, which is right for everything that is not compute-heavy." } }, type: 'object', @@ -1774,6 +1796,11 @@ export const NodeSourceSchema = { code: { type: 'string', title: 'Code' + }, + missing: { + type: 'boolean', + title: 'Missing', + default: false } }, type: 'object', @@ -2330,6 +2357,49 @@ export const RemoteUserBodySchema = { title: 'RemoteUserBody' } as const; +export const ResourcesSchema = { + properties: { + cpus: { + type: 'integer', + minimum: 1, + title: 'Cpus', + description: 'Cores held for the whole execution. Also what the thread-pool variables are set to, so a library sizing itself to the machine sizes itself to this instead.', + default: 1 + }, + gpus: { + type: 'integer', + minimum: 0, + title: 'Gpus', + description: 'Whole devices held for the whole execution, named to the node through CUDA_VISIBLE_DEVICES. Nothing else is given them while it runs, which is what keeps two preallocating processes apart.', + default: 0 + }, + env: { + additionalProperties: { + type: 'string' + }, + type: 'object', + title: 'Env', + description: "Extra environment for the worker this node runs in, applied over what the allocation derives. Where a library's own tuning goes — XLA_FLAGS, XLA_PYTHON_CLIENT_MEM_FRACTION — since those are composed strings the engine must not invent." + } + }, + additionalProperties: false, + type: 'object', + title: 'Resources', + description: `What one execution of a node needs to have to itself. + +Declaring nothing is the default and means what it always did: the node +runs on the shared worker pool and nothing is accounted for it. That is +right for the kind of node most flows are made of — a poll, a threshold, a +message on its way somewhere. + +It is wrong for the other kind. A numerical library sizes its thread pool +to every core it can see, so a handful of them at once oversubscribe the +machine badly enough to starve the engine's own event loop, and a GPU +library that preallocates most of the card deadlocks when a second one +arrives. Both are a node saying how much of the machine it takes, which is +what this is.` +} as const; + export const RuleSchema = { properties: { events: { @@ -2390,6 +2460,18 @@ export const RunCreateSchema = { enum: ['api', 'cli', 'sdk'], title: 'Cause', default: 'api' + }, + idempotency_key: { + anyOf: [ + { + type: 'string', + maxLength: 64 + }, + { + type: 'null' + } + ], + title: 'Idempotency Key' } }, type: 'object', @@ -2432,6 +2514,11 @@ export const RunDetailSchema = { title: 'Origin Commit', default: '' }, + code_digest: { + type: 'string', + title: 'Code Digest', + default: '' + }, commit: { type: 'string', title: 'Commit', @@ -2765,6 +2852,18 @@ export const SweepEntrySchema = { } ], title: 'Seed' + }, + idempotency_key: { + anyOf: [ + { + type: 'string', + maxLength: 64 + }, + { + type: 'null' + } + ], + title: 'Idempotency Key' } }, type: 'object', @@ -3139,7 +3238,7 @@ export const ValidationIssueSchema = { properties: { code: { type: 'string', - enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial'], + enum: ['cycle', 'unconnected_input', 'missing_initial_value', 'node_error', 'unauthenticated_hook', 'self_loop_needs_initial', 'missing_source'], title: 'Code' }, message: { @@ -3526,6 +3625,11 @@ export const fluksio__api__routes__runs__RunRowSchema = { title: 'Origin Commit', default: '' }, + code_digest: { + type: 'string', + title: 'Code Digest', + default: '' + }, commit: { type: 'string', title: 'Commit', diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 1f28b75..a9103c9 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsReadRunsData, RunsReadRunsResponse, RunsReadOverviewResponse, RunsReadRunData, RunsReadRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsReadRunsData, RunsReadRunsResponse, RunsReadOverviewResponse, RunsReadRunData, RunsReadRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AlertsService { /** @@ -2248,6 +2248,23 @@ export class WorkersService { }); } + /** + * Read Resources + * What this machine has free, and which nodes are queued for it. + * + * A node waiting its turn looks exactly like a node that has hung — the run + * sits at `running` and says nothing — so what is waiting, and for what, has + * to be readable somewhere. + * @returns unknown Successful Response + * @throws ApiError + */ + public static readResources(): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/workers/resources' + }); + } + /** * Issue Token * Mint the credential a worker presents when it dials in. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 0f53e53..d011026 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -425,6 +425,7 @@ export type fluksio__api__routes__runs__RunRow = { }; params_digest: string; origin_commit?: string; + code_digest?: string; commit?: string; seed: (number | null); group_id: (string | null); @@ -636,6 +637,10 @@ export type NodeDef_Input = { * Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own. */ cache?: boolean; + /** + * What one execution of this node holds while it runs. Absent — the default — means it is not accounted for and shares the engine's workers, which is right for everything that is not compute-heavy. + */ + resources?: (Resources | null); }; /** @@ -676,6 +681,10 @@ export type NodeDef_Output = { * Whether a batch run may reuse an earlier execution of this node with the same source, settings and inputs. Turn it off for a function whose answer can change on its own. */ cache?: boolean; + /** + * What one execution of this node holds while it runs. Absent — the default — means it is not accounted for and shares the engine's workers, which is right for everything that is not compute-heavy. + */ + resources?: (Resources | null); }; /** @@ -683,6 +692,7 @@ export type NodeDef_Output = { */ export type NodeSource = { code: string; + missing?: boolean; }; /** @@ -849,6 +859,38 @@ export type RemoteUserBody = { code: string; }; +/** + * What one execution of a node needs to have to itself. + * + * Declaring nothing is the default and means what it always did: the node + * runs on the shared worker pool and nothing is accounted for it. That is + * right for the kind of node most flows are made of — a poll, a threshold, a + * message on its way somewhere. + * + * It is wrong for the other kind. A numerical library sizes its thread pool + * to every core it can see, so a handful of them at once oversubscribe the + * machine badly enough to starve the engine's own event loop, and a GPU + * library that preallocates most of the card deadlocks when a second one + * arrives. Both are a node saying how much of the machine it takes, which is + * what this is. + */ +export type Resources = { + /** + * Cores held for the whole execution. Also what the thread-pool variables are set to, so a library sizing itself to the machine sizes itself to this instead. + */ + cpus?: number; + /** + * Whole devices held for the whole execution, named to the node through CUDA_VISIBLE_DEVICES. Nothing else is given them while it runs, which is what keeps two preallocating processes apart. + */ + gpus?: number; + /** + * Extra environment for the worker this node runs in, applied over what the allocation derives. Where a library's own tuning goes — XLA_FLAGS, XLA_PYTHON_CLIENT_MEM_FRACTION — since those are composed strings the engine must not invent. + */ + env?: { + [key: string]: (string); + }; +}; + /** * Which events go to which channels. */ @@ -866,6 +908,7 @@ export type RunCreate = { draft?: boolean; no_cache?: boolean; cause?: 'api' | 'cli' | 'sdk'; + idempotency_key?: (string | null); }; export type cause = 'api' | 'cli' | 'sdk'; @@ -881,6 +924,7 @@ export type RunDetail = { }; params_digest: string; origin_commit?: string; + code_digest?: string; commit?: string; seed: (number | null); group_id: (string | null); @@ -979,6 +1023,7 @@ export type SweepEntry = { [key: string]: unknown; }; seed?: (number | null); + idempotency_key?: (string | null); }; export type Token = { @@ -1063,7 +1108,7 @@ export type ValidationError = { * Something wrong with a flow — a fault, or merely advisory. */ export type ValidationIssue = { - code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial'; + code: 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source'; message: string; flow?: string; nodes?: Array<(string)>; @@ -1076,7 +1121,7 @@ export type ValidationIssue = { readonly advisory: boolean; }; -export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial'; +export type code = 'cycle' | 'unconnected_input' | 'missing_initial_value' | 'node_error' | 'unauthenticated_hook' | 'self_loop_needs_initial' | 'missing_source'; export type ValidationResult = { issues?: Array; @@ -1707,6 +1752,8 @@ export type UtilsHealthResponse = ({ export type WorkersReadWorkersResponse = (Array); +export type WorkersReadResourcesResponse = (unknown); + export type WorkersIssueTokenData = { requestBody: TokenRequest; };