diff --git a/.github/workflows/amiga.yml b/.github/workflows/amiga.yml new file mode 100644 index 0000000..4e99514 --- /dev/null +++ b/.github/workflows/amiga.yml @@ -0,0 +1,43 @@ +name: Amiga build + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + amiga-build: + runs-on: ubuntu-latest + env: + AMIGA_DEV_IMAGE: ghcr.io/ploos-as/amiga-dev:edge + AMIGA_CPU_PROFILE: "68000" + steps: + - uses: actions/checkout@v4 + - name: Pull amiga-dev + run: docker pull "$AMIGA_DEV_IMAGE" + - name: Verify amiga-dev contract + run: docker run --rm "$AMIGA_DEV_IMAGE" amiga-check + - name: Build and stage AmiRender Amiga client + run: | + mkdir -p amiga-runtime-payload/C amiga-runtime-payload/S + chmod 0777 amiga-runtime-payload/C + cp clients/amiga/runtime/amiga-runtime.json amiga-runtime-payload/ + cp clients/amiga/runtime/run-test amiga-runtime-payload/S/ + docker run --rm \ + -e AMIGA_CPU_PROFILE="$AMIGA_CPU_PROFILE" \ + -v "$PWD:/workspace:ro" \ + -v "$PWD/amiga-runtime-payload/C:/out" \ + "$AMIGA_DEV_IMAGE" \ + amiga-build sh -ec 'cd /workspace/clients/amiga && make TARGET=/out/amirender CC=m68k-amigaos-gcc CFLAGS="$AMIGA_CPU_FLAGS -Os -Wall -Wextra -Werror" && test -s /out/amirender && amiga-inspect /out/amirender' + test -s amiga-runtime-payload/C/amirender + - name: Upload classic runtime payload + uses: actions/upload-artifact@v4 + with: + name: amirender-m1-amiga-runtime-payload + path: amiga-runtime-payload/ + if-no-files-found: error + +# M1 uses :edge as the consumer-integration gate. Pin to a qualified immutable +# amiga-dev sha-* tag or digest before release qualification. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e99e445..817ce39 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -3,14 +3,41 @@ on: push: pull_request: jobs: - m0: + m1: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-go@v5 with: go-version: '1.22.x' - - run: gofmt -w cmd + - name: Install POV-Ray + run: | + sudo apt-get update + sudo DEBIAN_FRONTEND=noninteractive apt-get install -y povray + - run: gofmt -w cmd internal - run: git diff --exit-code - run: go test ./... - run: go build ./cmd/amirender-node ./cmd/amirender-controller + - name: Test portable Amiga client + run: | + cc -std=c99 -Wall -Wextra -Werror -pedantic \ + -Iclients/amiga/include \ + clients/amiga/src/submit.c clients/amiga/tests/test_submit.c \ + -o build-amiga-client-test + ./build-amiga-client-test + cc -std=c99 -Wall -Wextra -Werror -pedantic \ + -Iclients/amiga/include \ + clients/amiga/src/upload.c clients/amiga/tests/test_upload.c \ + -o build-amiga-upload-test + ./build-amiga-upload-test + cc -std=c99 -Wall -Wextra -Werror -pedantic \ + -Iclients/amiga/include \ + clients/amiga/src/submit.c clients/amiga/src/upload.c clients/amiga/src/client.c \ + clients/amiga/tests/test_client.c \ + -o build-amiga-transport-test + ./build-amiga-transport-test + - name: Render POV-Ray smoke scene + run: | + mkdir -p build + povray +Itests/scenes/m1-smoke.pov +Obuild/m1-smoke.png +W160 +H120 -D +FN + test -s build/m1-smoke.png diff --git a/README.md b/README.md index 0779140..60948a1 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # AmiRender -A render accelerator appliance and distributed rendering system for classic Amiga workflows. +A render accelerator appliance, distributed Linux render farm, and preservation bridge for classic 3D workflows. The Amiga is the workstation. AmiRender provides the modern compute. @@ -28,6 +28,36 @@ AmiRender is intended to support, over time: - AmiTerrain - Ami3D - Blender/Cycles/Eevee as appliance/farm backends +- Atari ST CAD-3D / Cyber Studio family +- Freescape / 3D Construction Kit scenes from C64, Atari ST, and Amiga + +## Render modes + +AmiRender keeps preservation and enhancement as separate, explicit goals: + +- `original`: use the original renderer/software where practical, on original hardware or a qualified emulator/VM, preserving native scene and output formats. +- `authentic`: reproduce the original platform renderer, palette, resolution, shading, and relevant limitations in an automatable AmiRender backend. +- `enhanced`: preserve the source scene semantics in AmiRender IR and render with modern backends for the best practical result. + +Importers must preserve original values and metadata alongside normalized IR data. Import must not destructively "improve" source assets. + +Native formats are first-class preservation inputs/outputs. Planned examples include Atari CAD-3D/Cyber formats and Freescape/3D Construction Kit data. Modern interchange formats are additional outputs, not replacements for native-format support. + +## Render farm + +POV-Ray and Blender are first-class modern Linux farm backends. The scheduler is intended to support amd64 and ARM64 CPU workers and GPU-capable workers where the backend supports them. Retro scene import is independent of the renderer: a preserved classic scene may be rendered in original/authentic mode or sent through the same modern farm in enhanced mode. + +## Landscape roadmap + +Landscape generation is a first-class render source, with AmiTerrain owning terrain semantics and format compatibility. + +- Preserve Vista/VistaPro, Scenery Animator, World Construction Set and other supported classic landscape inputs through AmiTerrain. +- Render preserved landscape scenes in `original`, `authentic`, or `enhanced` mode. +- Accept AmiTerrain Terrain IR without discarding native source metadata. +- Use POV-Ray and Blender/Cycles/Eevee as modern Linux farm targets for enhanced landscape rendering. +- Support modern open-source terrain workflows through AmiTerrain interchange, including TerraForge3D, Blender terrain/erosion workflows, generic heightmaps, meshes and DEM/GIS-derived terrain. +- Keep reverse conversion possible where representable: modern terrain may be reduced/quantized into formats usable by original classic landscape software. +- AmiRender does not replace AmiTerrain's importers or generators; it schedules and renders their outputs. ## User experience diff --git a/clients/amiga/Makefile b/clients/amiga/Makefile new file mode 100644 index 0000000..9bbc1cf --- /dev/null +++ b/clients/amiga/Makefile @@ -0,0 +1,21 @@ +CC = m68k-amigaos-gcc +CFLAGS ?= -O2 -Wall -Wextra +CPPFLAGS ?= -Iinclude +TARGET ?= amirender + +SOURCES = \ + src/submit.c \ + src/upload.c \ + src/client.c \ + src/bsdsocket_amiga.c \ + src/main_amiga.c + +.PHONY: all clean + +all: $(TARGET) + +$(TARGET): $(SOURCES) + $(CC) $(CPPFLAGS) $(CFLAGS) $(SOURCES) -o $@ + +clean: + rm -f $(TARGET) diff --git a/clients/amiga/README.md b/clients/amiga/README.md new file mode 100644 index 0000000..1c87835 --- /dev/null +++ b/clients/amiga/README.md @@ -0,0 +1,25 @@ +# AmiRender Amiga client + +This directory contains the first AmigaOS/m68k submission client components. + +M1 deliberately separates the portable wire-protocol builder from networking. The +portable C layer builds the same newline-delimited JSON `SUBMIT` message accepted +by `amirender-node`. The Amiga transport layer will use `bsdsocket.library`. + +The protocol builder rejects control characters, quotes and backslashes in text +fields rather than attempting incomplete JSON escaping. A later protocol utility +may add full escaping. + +Target integration: + +```text +Amiga application / CLI / ARexx + | + portable submit builder + | + bsdsocket.library + | + AmiRender node +``` + +No proprietary AmigaOS or Kickstart files belong in this repository. diff --git a/clients/amiga/docs/BUILD-AMIGA.md b/clients/amiga/docs/BUILD-AMIGA.md new file mode 100644 index 0000000..71a70b8 --- /dev/null +++ b/clients/amiga/docs/BUILD-AMIGA.md @@ -0,0 +1,29 @@ +# Building the AmigaOS/m68k client + +The portable client code is tested by the normal Linux CI. The Amiga-specific +transport lives in `src/bsdsocket_amiga.c` and targets an AmigaOS-compatible +m68k toolchain with `bsdsocket.library` development headers. + +Expected source set: + +```text +src/submit.c +src/client.c +src/bsdsocket_amiga.c +``` + +The transport opens `bsdsocket.library` version 4, resolves the controller or +worker host with `gethostbyname()`, creates an IPv4 TCP socket and exposes it +through the portable `amirender_transport` interface. + +The default AmiRender TCP port is 6800. + +## Qualification + +M1 qualification should compile this source with the Ploos `amiga-dev` Bebbo +toolchain, then execute the resulting m68k client in `amiga-runtime` against a +real AmiRender node. Runtime qualification must use an m68k Amiga environment; +AROS/i386 is not a substitute for this gate. + +No Kickstart ROM, AmigaOS installation, or other proprietary system file is +stored in this repository. diff --git a/clients/amiga/include/amirender_bsdsocket.h b/clients/amiga/include/amirender_bsdsocket.h new file mode 100644 index 0000000..744ac4a --- /dev/null +++ b/clients/amiga/include/amirender_bsdsocket.h @@ -0,0 +1,18 @@ +#ifndef AMIRENDER_BSDSOCKET_H +#define AMIRENDER_BSDSOCKET_H + +#include "amirender_transport.h" + +struct amirender_bsdsocket { + int socket_fd; +}; + +int amirender_bsdsocket_connect( + struct amirender_bsdsocket *socket, + const char *host, + unsigned short port, + struct amirender_transport *transport); + +void amirender_bsdsocket_close(struct amirender_bsdsocket *socket); + +#endif diff --git a/clients/amiga/include/amirender_submit.h b/clients/amiga/include/amirender_submit.h new file mode 100644 index 0000000..568f0b7 --- /dev/null +++ b/clients/amiga/include/amirender_submit.h @@ -0,0 +1,18 @@ +#ifndef AMIRENDER_SUBMIT_H +#define AMIRENDER_SUBMIT_H + +#include + +struct amirender_job { + const char *id; + const char *engine; + const char *scene; + const char *output; + int frame; + int width; + int height; +}; + +int amirender_build_submit(char *buffer, size_t size, const struct amirender_job *job); + +#endif diff --git a/clients/amiga/include/amirender_transport.h b/clients/amiga/include/amirender_transport.h new file mode 100644 index 0000000..281e9c1 --- /dev/null +++ b/clients/amiga/include/amirender_transport.h @@ -0,0 +1,70 @@ +#ifndef AMIRENDER_TRANSPORT_H +#define AMIRENDER_TRANSPORT_H + +#include + +struct amirender_job; + +struct amirender_transport { + void *context; + int (*send)(void *context, const char *data, size_t length); + int (*receive)(void *context, char *buffer, size_t size); +}; + +int amirender_upload_asset( + struct amirender_transport *transport, + const char *name, + const unsigned char *data, + size_t data_size, + char *asset, + size_t asset_size); + +int amirender_download_asset( + struct amirender_transport *transport, + const char *asset, + unsigned char *data, + size_t data_size, + size_t *received_size); + +int amirender_upload_begin( + struct amirender_transport *transport, + const char *name, + size_t total_size, + char *asset, + size_t asset_size); + +int amirender_upload_chunk( + struct amirender_transport *transport, + const char *asset, + size_t offset, + const unsigned char *data, + size_t data_size); + +int amirender_upload_end( + struct amirender_transport *transport, + const char *asset, + size_t total_size); + +int amirender_download_chunk( + struct amirender_transport *transport, + const char *asset, + size_t offset, + unsigned char *data, + size_t data_size, + size_t *received_size, + int *eof); + +int amirender_submit_job( + struct amirender_transport *transport, + const struct amirender_job *job, + char *reply, + size_t reply_size); + +int amirender_extract_output_asset( + const char *reply, char *asset, size_t asset_size); + +int amirender_download_file( + struct amirender_transport *transport, const char *asset, + const char *path, size_t *total_size); + +#endif diff --git a/clients/amiga/include/amirender_upload.h b/clients/amiga/include/amirender_upload.h new file mode 100644 index 0000000..277e5c2 --- /dev/null +++ b/clients/amiga/include/amirender_upload.h @@ -0,0 +1,13 @@ +#ifndef AMIRENDER_UPLOAD_H +#define AMIRENDER_UPLOAD_H + +#include + +int amirender_build_upload( + char *buffer, + size_t size, + const char *name, + const unsigned char *data, + size_t data_size); + +#endif diff --git a/clients/amiga/runtime/amiga-runtime.json b/clients/amiga/runtime/amiga-runtime.json new file mode 100644 index 0000000..4ea619e --- /dev/null +++ b/clients/amiga/runtime/amiga-runtime.json @@ -0,0 +1,25 @@ +{ + "schema": 1, + "project": "AmiRender", + "milestone": "M1-amiga-client", + "architecture": "m68k", + "cpu": "68000", + "os": "AmigaOS", + "emulator_profiles": ["a500-os204", "a500plus-os2", "a1200-020-os3"], + "payload": { + "programs": ["C/amirender"], + "script": "S/run-test" + }, + "result": { + "logs": ["T/amirender.log"], + "required_lines": ["AMIRENDER_EXEC=PASS", "AMIRENDER_USAGE_RC=20"] + }, + "policy": { + "roms_in_repository": false, + "amigaos_files_in_repository": false + }, + "requirements": { + "network": false, + "bsdsocket_library": true + } +} diff --git a/clients/amiga/runtime/run-test b/clients/amiga/runtime/run-test new file mode 100644 index 0000000..84b9870 --- /dev/null +++ b/clients/amiga/runtime/run-test @@ -0,0 +1,14 @@ +; AmiRender M1 classic runtime smoke test +; The client is intentionally invoked without arguments. +; Exit code 20 proves the m68k executable reached its CLI validation path. + +C:amirender >T:amirender-cli.out +SET rc $RC + +ECHO "AMIRENDER_USAGE_RC=$rc" >T:amirender.log + +IF $rc EQ 20 + ECHO "AMIRENDER_EXEC=PASS" >>T:amirender.log +ELSE + ECHO "AMIRENDER_EXEC=FAIL" >>T:amirender.log +ENDIF diff --git a/clients/amiga/src/bsdsocket_amiga.c b/clients/amiga/src/bsdsocket_amiga.c new file mode 100644 index 0000000..3fc2417 --- /dev/null +++ b/clients/amiga/src/bsdsocket_amiga.c @@ -0,0 +1,166 @@ +/* + * AmigaOS/m68k transport for AmiRender. + * + * This translation unit is intentionally not part of the host build. It is + * compiled by the Amiga/Bebbo toolchain where bsdsocket.library headers and + * stubs are available. + */ + +#include "amirender_bsdsocket.h" + +#include +#include +#include + +#include +#include +#include + +struct Library *SocketBase = NULL; + + +static int parse_ipv4(const char *text, unsigned long *result) +{ + unsigned long value = 0; + int part; + + if (text == NULL || result == NULL) { + return -1; + } + + for (part = 0; part < 4; ++part) { + unsigned long octet = 0; + int digits = 0; + + while (*text >= '0' && *text <= '9') { + octet = octet * 10UL + (unsigned long)(*text - '0'); + if (octet > 255UL) { + return -1; + } + ++text; + ++digits; + } + if (digits == 0) { + return -1; + } + value = (value << 8) | octet; + if (part < 3) { + if (*text != '.') { + return -1; + } + ++text; + } + } + if (*text != '\0') { + return -1; + } + + *result = value; + return 0; +} + +static int socket_send(void *context, const char *data, size_t length) +{ + struct amirender_bsdsocket *state = (struct amirender_bsdsocket *)context; + size_t sent = 0; + + while (sent < length) { + int n = send(state->socket_fd, (void *)(data + sent), length - sent, 0); + if (n <= 0) { + return -1; + } + sent += (size_t)n; + } + return (int)sent; +} + +static int socket_receive(void *context, char *buffer, size_t size) +{ + struct amirender_bsdsocket *state = (struct amirender_bsdsocket *)context; + size_t received = 0; + + if (buffer == NULL || size == 0) { + return -1; + } + + while (received < size) { + int n = recv(state->socket_fd, buffer + received, size - received, 0); + size_t i; + + if (n <= 0) { + return received > 0 ? (int)received : -1; + } + + for (i = 0; i < (size_t)n; ++i) { + if (buffer[received + i] == '\n') { + return (int)(received + i + 1); + } + } + received += (size_t)n; + } + + return (int)received; +} + +int amirender_bsdsocket_connect( + struct amirender_bsdsocket *state, + const char *host, + unsigned short port, + struct amirender_transport *transport) +{ + struct sockaddr_in address; + int fd; + unsigned long ipv4; + + if (state == NULL || host == NULL || transport == NULL) { + return -1; + } + + SocketBase = OpenLibrary("bsdsocket.library", 4); + if (SocketBase == NULL) { + return -1; + } + + fd = socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + CloseLibrary(SocketBase); + SocketBase = NULL; + return -1; + } + + memset(&address, 0, sizeof(address)); + address.sin_family = AF_INET; + address.sin_port = htons(port); + if (parse_ipv4(host, &ipv4) != 0) { + CloseSocket(fd); + CloseLibrary(SocketBase); + SocketBase = NULL; + return -1; + } + address.sin_addr.s_addr = htonl(ipv4); + + if (connect(fd, (struct sockaddr *)&address, sizeof(address)) < 0) { + CloseSocket(fd); + CloseLibrary(SocketBase); + SocketBase = NULL; + return -1; + } + + state->socket_fd = fd; + transport->context = state; + transport->send = socket_send; + transport->receive = socket_receive; + return 0; +} + +void amirender_bsdsocket_close(struct amirender_bsdsocket *state) +{ + if (state != NULL && state->socket_fd >= 0) { + CloseSocket(state->socket_fd); + state->socket_fd = -1; + } + if (SocketBase != NULL) { + CloseLibrary(SocketBase); + SocketBase = NULL; + } +} diff --git a/clients/amiga/src/client.c b/clients/amiga/src/client.c new file mode 100644 index 0000000..474c0e7 --- /dev/null +++ b/clients/amiga/src/client.c @@ -0,0 +1,468 @@ +#include "amirender_submit.h" +#include "amirender_transport.h" +#include "amirender_upload.h" + +#include + +#ifdef __m68k__ +#include +#endif + +#define AMIRENDER_FILE_CHUNK_SIZE 2048 +#define AMIRENDER_FILE_PATH_SIZE 1024 +#include + +int amirender_upload_asset( + struct amirender_transport *transport, + const char *name, + const unsigned char *data, + size_t data_size, + char *asset, + size_t asset_size) +{ + char request[8192]; + char reply[1024]; + const char *key = "\"asset\":\""; + const char *start; + const char *end; + size_t length; + int request_length; + int received; + + if (transport == NULL || transport->send == NULL || transport->receive == NULL || + asset == NULL || asset_size < 2) { + return -1; + } + request_length = amirender_build_upload(request, sizeof(request), name, data, data_size); + if (request_length < 0) { + return -1; + } + if (transport->send(transport->context, request, (size_t)request_length) != request_length) { + return -1; + } + received = transport->receive(transport->context, reply, sizeof(reply) - 1); + if (received <= 0 || (size_t)received >= sizeof(reply)) { + return -1; + } + reply[received] = '\0'; + if (strstr(reply, "\"type\":\"STAGED\"") == NULL) { + return -1; + } + start = strstr(reply, key); + if (start == NULL) { + return -1; + } + start += strlen(key); + end = strchr(start, '"'); + if (end == NULL) { + return -1; + } + length = (size_t)(end - start); + if (length == 0 || length >= asset_size) { + return -1; + } + memcpy(asset, start, length); + asset[length] = '\0'; + return 0; +} + +static int base64_value(char ch) +{ + if (ch >= 'A' && ch <= 'Z') return ch - 'A'; + if (ch >= 'a' && ch <= 'z') return ch - 'a' + 26; + if (ch >= '0' && ch <= '9') return ch - '0' + 52; + if (ch == '+') return 62; + if (ch == '/') return 63; + return -1; +} + +int amirender_download_asset( + struct amirender_transport *transport, + const char *asset, + unsigned char *data, + size_t data_size, + size_t *received_size) +{ + char request[1200]; + char reply[8192]; + const char *key = "\"data\":\""; + const char *p; + const char *end; + size_t out = 0; + int received; + + if (transport == NULL || transport->send == NULL || transport->receive == NULL || + asset == NULL || data == NULL || received_size == NULL) return -1; + if (strchr(asset, '\"') != NULL || strchr(asset, '\\') != NULL) return -1; + if (snprintf(request, sizeof(request), "{\"type\":\"DOWNLOAD\",\"asset\":\"%s\"}\n", asset) < 0 || + strlen(request) >= sizeof(request)) return -1; + if (transport->send(transport->context, request, strlen(request)) != (int)strlen(request)) return -1; + received = transport->receive(transport->context, reply, sizeof(reply) - 1); + if (received <= 0 || (size_t)received >= sizeof(reply)) return -1; + reply[received] = '\0'; + if (strstr(reply, "\"type\":\"DATA\"") == NULL) return -1; + p = strstr(reply, key); + if (p == NULL) return -1; + p += strlen(key); + end = strchr(p, '\"'); + if (end == NULL) return -1; + + while (p < end) { + int a, b, d, e; + unsigned int v; + if (end - p < 4) return -1; + a = base64_value(p[0]); b = base64_value(p[1]); + d = p[2] == '=' ? 0 : base64_value(p[2]); + e = p[3] == '=' ? 0 : base64_value(p[3]); + if (a < 0 || b < 0 || d < 0 || e < 0) return -1; + if (p[2] == '=' && p[3] != '=') return -1; + if ((p[2] == '=' || p[3] == '=') && p + 4 != end) return -1; + if (p[2] == '=' && (b & 15) != 0) return -1; + if (p[3] == '=' && p[2] != '=' && (d & 3) != 0) return -1; + v = ((unsigned int)a << 18) | ((unsigned int)b << 12) | + ((unsigned int)d << 6) | (unsigned int)e; + if (out >= data_size) return -1; + data[out++] = (unsigned char)(v >> 16); + if (p[2] != '=') { + if (out >= data_size) return -1; + data[out++] = (unsigned char)(v >> 8); + } + if (p[3] != '=') { + if (out >= data_size) return -1; + data[out++] = (unsigned char)v; + } + p += 4; + } + *received_size = out; + return 0; +} + + +static int extract_string_field(const char *reply, const char *key, char *out, size_t out_size) +{ + const char *start = strstr(reply, key); + const char *end; + size_t length; + if (start == NULL || out == NULL || out_size < 2) return -1; + start += strlen(key); + end = strchr(start, '"'); + if (end == NULL) return -1; + length = (size_t)(end - start); + if (length == 0 || length >= out_size) return -1; + memcpy(out, start, length); + out[length] = '\0'; + return 0; +} + +int amirender_upload_begin( + struct amirender_transport *transport, const char *name, size_t total_size, + char *asset, size_t asset_size) +{ + char request[512]; + char reply[1024]; + int n, received; + if (transport == NULL || transport->send == NULL || transport->receive == NULL || + name == NULL || asset == NULL || total_size == 0) return -1; + n = snprintf(request, sizeof(request), + "{\"type\":\"UPLOAD_BEGIN\",\"name\":\"%s\",\"size\":%lu}\n", + name, (unsigned long)total_size); + if (n < 0 || (size_t)n >= sizeof(request)) return -1; + if (transport->send(transport->context, request, (size_t)n) != n) return -1; + received = transport->receive(transport->context, reply, sizeof(reply) - 1); + if (received <= 0 || (size_t)received >= sizeof(reply)) return -1; + reply[received] = '\0'; + if (strstr(reply, "\"type\":\"STAGING\"") == NULL) return -1; + if (strstr(reply, "\"id\":\"") != NULL && + extract_string_field(reply, "\"id\":\"", asset, asset_size) == 0) { + return 0; + } + return extract_string_field(reply, "\"asset\":\"", asset, asset_size); +} + +int amirender_upload_chunk( + struct amirender_transport *transport, const char *asset, size_t offset, + const unsigned char *data, size_t data_size) +{ + char encoded[2733]; + char request[3584]; + char reply[1024]; + static const char alphabet[] = + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + size_t i, out = 0; + int n, received; + if (transport == NULL || transport->send == NULL || transport->receive == NULL || + asset == NULL || data == NULL || data_size == 0 || data_size > 2048) return -1; + if (strchr(asset, '"') != NULL || strchr(asset, '\\') != NULL) return -1; + for (i = 0; i < data_size; i += 3) { + unsigned int v = (unsigned int)data[i] << 16; + v |= i + 1 < data_size ? (unsigned int)data[i + 1] << 8 : 0; + v |= i + 2 < data_size ? (unsigned int)data[i + 2] : 0; + encoded[out++] = alphabet[(v >> 18) & 63]; + encoded[out++] = alphabet[(v >> 12) & 63]; + encoded[out++] = i + 1 < data_size ? alphabet[(v >> 6) & 63] : '='; + encoded[out++] = i + 2 < data_size ? alphabet[v & 63] : '='; + } + encoded[out] = '\0'; + n = snprintf(request, sizeof(request), + "{\"type\":\"UPLOAD_CHUNK\",\"asset\":\"%s\",\"offset\":%lu,\"data\":\"%s\"}\n", + asset, (unsigned long)offset, encoded); + if (n < 0 || (size_t)n >= sizeof(request)) return -1; + if (transport->send(transport->context, request, (size_t)n) != n) return -1; + received = transport->receive(transport->context, reply, sizeof(reply) - 1); + if (received <= 0 || (size_t)received >= sizeof(reply)) return -1; + reply[received] = '\0'; + return strstr(reply, "\"type\":\"CHUNKED\"") != NULL ? 0 : -1; +} + +int amirender_upload_end( + struct amirender_transport *transport, const char *asset, size_t total_size) +{ + char request[1200]; + char reply[1024]; + int n, received; + if (transport == NULL || transport->send == NULL || transport->receive == NULL || + asset == NULL || total_size == 0) return -1; + if (strchr(asset, '"') != NULL || strchr(asset, '\\') != NULL) return -1; + n = snprintf(request, sizeof(request), + "{\"type\":\"UPLOAD_END\",\"asset\":\"%s\",\"size\":%lu}\n", + asset, (unsigned long)total_size); + if (n < 0 || (size_t)n >= sizeof(request)) return -1; + if (transport->send(transport->context, request, (size_t)n) != n) return -1; + received = transport->receive(transport->context, reply, sizeof(reply) - 1); + if (received <= 0 || (size_t)received >= sizeof(reply)) return -1; + reply[received] = '\0'; + return strstr(reply, "\"type\":\"STAGED\"") != NULL ? 0 : -1; +} + +/* Return a top-level JSON field, ignoring quoted string contents. */ +static const char *json_field(const char *json, const char *key, int *matches) +{ + const char *p = json; + const char *found = NULL; + size_t len = strlen(key); + int depth = 0, quoted = 0, escaped = 0, closed = 0; + *matches = 0; + if (*p != '{') return NULL; + while (*p) { + if (closed) { + if (*p != ' ' && *p != '\\n' && *p != '\\r' && *p != '\\t') return NULL; + ++p; + continue; + } + if (quoted) { + if (escaped) { + if (*p == 'u') { + int i; + for (i = 1; i <= 4; ++i) { + char h = p[i]; + if (!((h >= '0' && h <= '9') || + (h >= 'a' && h <= 'f') || + (h >= 'A' && h <= 'F'))) return NULL; + } + p += 4; + } else if (*p != '"' && *p != '\\' && *p != '/' && + *p != 'b' && *p != 'f' && *p != 'n' && + *p != 'r' && *p != 't') return NULL; + escaped = 0; + } + else if (*p == '\\') escaped = 1; + else if ((unsigned char)*p < 0x20) return NULL; + else if (*p == '"') quoted = 0; + } else if (*p == '{') ++depth; + else if (*p == '}') { + if (--depth < 0) return NULL; + if (depth == 0) closed = 1; + } + else if (*p == '"' && depth == 1) { + if (strncmp(p + 1, key, len) == 0 && + p[len + 1] == '"' && p[len + 2] == ':') { + found = p + len + 3; + ++*matches; + } + quoted = 1; + } else if (*p == '"') quoted = 1; + ++p; + } + return closed && depth == 0 && !quoted ? found : NULL; +} + +int amirender_download_chunk( + struct amirender_transport *transport, const char *asset, size_t offset, + unsigned char *data, size_t data_size, size_t *received_size, int *eof) +{ + char request[1200]; + char reply[3584]; + const char *p, *end, *field; + int matches; + int is_eof; + size_t out = 0; + int n, received; + if (transport == NULL || transport->send == NULL || transport->receive == NULL || + asset == NULL || data == NULL || received_size == NULL || eof == NULL || + data_size == 0 || data_size > 2048) return -1; + if (strchr(asset, '"') != NULL || strchr(asset, '\\') != NULL) return -1; + n = snprintf(request, sizeof(request), + "{\"type\":\"DOWNLOAD_CHUNK\",\"asset\":\"%s\",\"offset\":%lu,\"size\":%lu}\n", + asset, (unsigned long)offset, (unsigned long)data_size); + if (n < 0 || (size_t)n >= sizeof(request)) return -1; + if (transport->send(transport->context, request, (size_t)n) != n) return -1; + received = transport->receive(transport->context, reply, sizeof(reply) - 1); + if (received <= 0 || (size_t)received >= sizeof(reply)) return -1; + reply[received] = '\0'; + field = json_field(reply, "type", &matches); + if (matches != 1 || field == NULL || + strncmp(field, "\"DATA\"", 6) != 0) return -1; + field = json_field(reply, "eof", &matches); + if (matches != 1 || field == NULL) return -1; + if (strncmp(field, "true", 4) == 0 && + (field[4] == ',' || field[4] == '}' || field[4] == ' ')) is_eof = 1; + else if (strncmp(field, "false", 5) == 0 && + (field[5] == ',' || field[5] == '}' || field[5] == ' ')) is_eof = 0; + else return -1; + field = json_field(reply, "data", &matches); + if (matches > 1) return -1; + if (field == NULL) { + if (matches == 0 && is_eof) { + *received_size = 0; *eof = 1; return 0; + } + return -1; + } + if (*field != '"') return -1; + p = field + 1; + end = strchr(p, '"'); + if (end == NULL) return -1; + while (p < end) { + int a, b, d, e; + unsigned int v; + if (end - p < 4) return -1; + a = base64_value(p[0]); b = base64_value(p[1]); + d = p[2] == '=' ? 0 : base64_value(p[2]); + e = p[3] == '=' ? 0 : base64_value(p[3]); + if (a < 0 || b < 0 || d < 0 || e < 0) return -1; + if (p[2] == '=' && p[3] != '=') return -1; + if ((p[2] == '=' || p[3] == '=') && p + 4 != end) return -1; + if (p[2] == '=' && (b & 15) != 0) return -1; + if (p[3] == '=' && p[2] != '=' && (d & 3) != 0) return -1; + v = ((unsigned int)a << 18) | ((unsigned int)b << 12) | + ((unsigned int)d << 6) | (unsigned int)e; + if (out >= data_size) return -1; + data[out++] = (unsigned char)(v >> 16); + if (p[2] != '=') { if (out >= data_size) return -1; data[out++] = (unsigned char)(v >> 8); } + if (p[3] != '=') { if (out >= data_size) return -1; data[out++] = (unsigned char)v; } + p += 4; + } + *received_size = out; + *eof = is_eof; + return 0; +} + +int amirender_submit_job( + struct amirender_transport *transport, + const struct amirender_job *job, + char *reply, + size_t reply_size) +{ + char request[1024]; + int length; + int received; + + if (transport == NULL || transport->send == NULL || transport->receive == NULL || + reply == NULL || reply_size < 2) { + return -1; + } + + length = amirender_build_submit(request, sizeof(request), job); + if (length < 0) { + return -1; + } + if (transport->send(transport->context, request, (size_t)length) != length) { + return -1; + } + + received = transport->receive(transport->context, reply, reply_size - 1); + if (received <= 0 || (size_t)received >= reply_size) { + return -1; + } + reply[received] = '\0'; + + if (strstr(reply, "\"type\":\"COMPLETE\"") == NULL) { + return -1; + } + return 0; +} + +int amirender_extract_output_asset( + const char *reply, char *asset, size_t asset_size) +{ + const char *key = "\"asset_id\":\""; + const char *start; + const char *end; + size_t length; + + if (reply == NULL || asset == NULL || asset_size < 2) return -1; + start = strstr(reply, key); + if (start == NULL || start[strlen(key)] == '"') { + key = "\"output\":\""; + start = strstr(reply, key); + } + if (start == NULL) return -1; + start += strlen(key); + end = strchr(start, '"'); + if (end == NULL) return -1; + length = (size_t)(end - start); + if (length == 0 || length >= asset_size) return -1; + memcpy(asset, start, length); + asset[length] = '\0'; + return 0; +} + +int amirender_download_file( + struct amirender_transport *transport, const char *asset, + const char *path, size_t *total_size) +{ + FILE *file; + char temporary[AMIRENDER_FILE_PATH_SIZE]; + unsigned char buffer[AMIRENDER_FILE_CHUNK_SIZE]; + size_t offset = 0, count; + int eof = 0; + int name_length; + + if (transport == NULL || asset == NULL || path == NULL || total_size == NULL) return -1; + name_length = snprintf(temporary, sizeof(temporary), "%s.part", path); + if (name_length < 0 || (size_t)name_length >= sizeof(temporary)) return -1; + file = fopen(temporary, "wb"); + if (file == NULL) return -1; + while (!eof) { + if (amirender_download_chunk( + transport, asset, offset, buffer, sizeof(buffer), &count, &eof) != 0) { + fclose(file); + remove(temporary); + return -1; + } + if (count > 0 && fwrite(buffer, 1, count, file) != count) { + fclose(file); + remove(temporary); + return -1; + } + offset += count; + if (count == 0 && (!eof || offset == 0)) { + fclose(file); + remove(temporary); + return -1; + } + } + if (fclose(file) != 0) { + remove(temporary); + return -1; + } +#ifdef __m68k__ + if (!Rename((STRPTR)temporary, (STRPTR)path)) { +#else + if (rename(temporary, path) != 0) { +#endif + remove(temporary); + return -1; + } + *total_size = offset; + return 0; +} diff --git a/clients/amiga/src/main_amiga.c b/clients/amiga/src/main_amiga.c new file mode 100644 index 0000000..fb41418 --- /dev/null +++ b/clients/amiga/src/main_amiga.c @@ -0,0 +1,148 @@ +#include "amirender_bsdsocket.h" +#include "amirender_submit.h" +#include "amirender_transport.h" + +#include +#include +#include + + +#define AMIRENDER_PORT 6800 +#define REPLY_SIZE 1024 +#define ASSET_SIZE 1024 +#define CHUNK_SIZE 2048 + +static const char *base_name(const char *path) +{ + const char *name = path; + const char *p; + + for (p = path; *p != '\0'; ++p) { + if (*p == '/' || *p == ':') { + name = p + 1; + } + } + return name; +} + +static int scene_size(const char *path, size_t *length) +{ + FILE *file; + long size; + + file = fopen(path, "rb"); + if (file == NULL) return -1; + if (fseek(file, 0, SEEK_END) != 0) { + fclose(file); + return -1; + } + size = ftell(file); + fclose(file); + if (size <= 0) return -1; + *length = (size_t)size; + return 0; +} + +static int upload_file( + struct amirender_transport *transport, const char *path, + char *asset, size_t asset_size) +{ + FILE *file; + unsigned char buffer[CHUNK_SIZE]; + size_t total, offset = 0, count; + + if (scene_size(path, &total) != 0) return -1; + if (amirender_upload_begin( + transport, base_name(path), total, asset, asset_size) != 0) return -1; + file = fopen(path, "rb"); + if (file == NULL) return -1; + while ((count = fread(buffer, 1, sizeof(buffer), file)) > 0) { + if (amirender_upload_chunk(transport, asset, offset, buffer, count) != 0) { + fclose(file); + return -1; + } + offset += count; + } + if (ferror(file) || offset != total) { + fclose(file); + return -1; + } + if (fclose(file) != 0) return -1; + return amirender_upload_end(transport, asset, total); +} + +static void usage(const char *program) +{ + fprintf(stderr, + "Usage: %s HOST SCENE OUTPUT [WIDTH HEIGHT]\n" + "Example: %s 192.168.1.50 DH0:Scenes/demo.pov RAM:frame.png 320 256\n", + program, program); +} + +int main(int argc, char **argv) +{ + struct amirender_bsdsocket socket_state = {-1}; + struct amirender_transport transport = {0}; + struct amirender_job job; + char reply[REPLY_SIZE]; + char asset[ASSET_SIZE]; + char output_asset[ASSET_SIZE]; + size_t output_size; + int width = 0; + int height = 0; + int rc; + + if (argc != 4 && argc != 6) { + usage(argv[0]); + return 20; + } + + if (argc == 6) { + width = atoi(argv[4]); + height = atoi(argv[5]); + if (width <= 0 || height <= 0) { + fprintf(stderr, "AmiRender: WIDTH and HEIGHT must be positive\n"); + return 20; + } + } + + job.id = "amiga-cli"; + job.engine = "povray"; + job.scene = argv[2]; + job.output = argv[3]; + job.frame = 0; + job.width = width; + job.height = height; + + if (amirender_bsdsocket_connect( + &socket_state, argv[1], AMIRENDER_PORT, &transport) != 0) { + fprintf(stderr, "AmiRender: cannot connect to %s:%d\n", + argv[1], AMIRENDER_PORT); + return 10; + } + + rc = upload_file(&transport, argv[2], asset, sizeof(asset)); + if (rc != 0) { + amirender_bsdsocket_close(&socket_state); + fprintf(stderr, "AmiRender: scene upload failed\n"); + return 10; + } + + job.scene = asset; + rc = amirender_submit_job(&transport, &job, reply, sizeof(reply)); + if (rc != 0 || amirender_extract_output_asset(reply, output_asset, sizeof(output_asset)) != 0) { + amirender_bsdsocket_close(&socket_state); + fprintf(stderr, "AmiRender: render failed\n"); + return 10; + } + + rc = amirender_download_file(&transport, output_asset, argv[3], &output_size); + amirender_bsdsocket_close(&socket_state); + if (rc != 0) { + fprintf(stderr, "AmiRender: output download failed\n"); + return 10; + } + + printf("AmiRender: wrote %lu bytes to %s\n", (unsigned long)output_size, argv[3]); + return 0; +} diff --git a/clients/amiga/src/submit.c b/clients/amiga/src/submit.c new file mode 100644 index 0000000..4f4c962 --- /dev/null +++ b/clients/amiga/src/submit.c @@ -0,0 +1,42 @@ +#include "amirender_submit.h" + +#include + +static int safe_text(const char *s) +{ + const unsigned char *p = (const unsigned char *)s; + + if (s == NULL) { + return 0; + } + while (*p != '\0') { + if (*p < 0x20 || *p == '"' || *p == '\\') { + return 0; + } + ++p; + } + return 1; +} + +int amirender_build_submit(char *buffer, size_t size, const struct amirender_job *job) +{ + int n; + + if (buffer == NULL || size == 0 || job == NULL || + !safe_text(job->id) || !safe_text(job->engine) || + !safe_text(job->scene) || !safe_text(job->output)) { + return -1; + } + + n = snprintf( + buffer, size, + "{\"type\":\"SUBMIT\",\"job\":{\"id\":\"%s\",\"engine\":\"%s\"," + "\"scene\":\"%s\",\"output\":\"%s\",\"frame\":%d,\"width\":%d,\"height\":%d}}\n", + job->id, job->engine, job->scene, job->output, + job->frame, job->width, job->height); + + if (n < 0 || (size_t)n >= size) { + return -1; + } + return n; +} diff --git a/clients/amiga/src/upload.c b/clients/amiga/src/upload.c new file mode 100644 index 0000000..3868420 --- /dev/null +++ b/clients/amiga/src/upload.c @@ -0,0 +1,71 @@ +#include "amirender_upload.h" + +#include +#include + +static const char base64_table[] = + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + +static int safe_name(const char *name) +{ + const unsigned char *p = (const unsigned char *)name; + + if (name == NULL || *name == '\0' || strcmp(name, ".") == 0 || strcmp(name, "..") == 0) { + return 0; + } + while (*p != '\0') { + if (*p < 0x20 || *p == '"' || *p == '\\' || *p == '/' || *p == ':') { + return 0; + } + ++p; + } + return 1; +} + +int amirender_build_upload( + char *buffer, + size_t size, + const char *name, + const unsigned char *data, + size_t data_size) +{ + size_t encoded_size; + size_t required; + size_t i; + size_t out; + int prefix; + + if (buffer == NULL || size == 0 || !safe_name(name) || data == NULL || data_size == 0) { + return -1; + } + if (data_size > (size_t)-1 - 2) { + return -1; + } + encoded_size = ((data_size + 2) / 3) * 4; + prefix = snprintf(buffer, size, "{\"type\":\"UPLOAD\",\"name\":\"%s\",\"data\":\"", name); + if (prefix < 0 || (size_t)prefix >= size) { + return -1; + } + required = (size_t)prefix + encoded_size + 3; + if (required > size) { + return -1; + } + + out = (size_t)prefix; + for (i = 0; i < data_size; i += 3) { + unsigned long v = (unsigned long)data[i] << 16; + size_t remain = data_size - i; + if (remain > 1) v |= (unsigned long)data[i + 1] << 8; + if (remain > 2) v |= data[i + 2]; + + buffer[out++] = base64_table[(v >> 18) & 63]; + buffer[out++] = base64_table[(v >> 12) & 63]; + buffer[out++] = remain > 1 ? base64_table[(v >> 6) & 63] : '='; + buffer[out++] = remain > 2 ? base64_table[v & 63] : '='; + } + buffer[out++] = '"'; + buffer[out++] = '}'; + buffer[out++] = '\n'; + buffer[out] = '\0'; + return (int)out; +} diff --git a/clients/amiga/tests/test_client.c b/clients/amiga/tests/test_client.c new file mode 100644 index 0000000..ebabd34 --- /dev/null +++ b/clients/amiga/tests/test_client.c @@ -0,0 +1,352 @@ +#include "amirender_submit.h" +#include "amirender_transport.h" + +#include +#include + +struct fake_transport { + char sent[1024]; + const char *reply; + const char *next_reply; + int receive_count; +}; + +static int fake_send(void *context, const char *data, size_t length) +{ + struct fake_transport *fake = (struct fake_transport *)context; + + if (length >= sizeof(fake->sent)) { + return -1; + } + memcpy(fake->sent, data, length); + fake->sent[length] = '\0'; + return (int)length; +} + +static int fake_receive(void *context, char *buffer, size_t size) +{ + struct fake_transport *fake = (struct fake_transport *)context; + const char *reply = fake->reply; + size_t length; + + if (fake->next_reply != NULL && fake->receive_count++ > 0) { + reply = fake->next_reply; + } + length = strlen(reply); + + if (length > size) { + return -1; + } + memcpy(buffer, reply, length); + return (int)length; +} + +int main(void) +{ + struct fake_transport fake = {{0}, NULL, NULL, 0}; + struct amirender_transport transport = {&fake, fake_send, fake_receive}; + struct amirender_job job = { + "amiga-0002", "povray", "DH0:Scenes/demo.pov", + "RAM:frame.png", 0, 320, 256 + }; + char reply[512]; + char asset[256]; + unsigned char downloaded[16]; + size_t downloaded_size = 0; + const unsigned char scene[] = "camera {}\n"; + + fake.reply = "{\"type\":\"COMPLETE\",\"job_id\":\"amiga-0002\"," + "\"engine\":\"povray\",\"output\":\"RAM:frame.png\"}\n"; + if (amirender_submit_job(&transport, &job, reply, sizeof(reply)) != 0) { + return 1; + } + if (strstr(fake.sent, "\"id\":\"amiga-0002\"") == NULL) { + fprintf(stderr, "job id missing from request\n"); + return 2; + } + if (strstr(reply, "\"type\":\"COMPLETE\"") == NULL) { + fprintf(stderr, "completion missing from reply\n"); + return 3; + } + + fake.reply = "{\"type\":\"STAGED\",\"asset\":\"/tmp/amirender/scene.pov\"}\n"; + if (amirender_upload_asset( + &transport, "scene.pov", scene, sizeof(scene) - 1, asset, sizeof(asset)) != 0) { + return 4; + } + if (strcmp(asset, "/tmp/amirender/scene.pov") != 0) { + fprintf(stderr, "unexpected staged asset: %s\n", asset); + return 5; + } + if (strstr(fake.sent, "\"type\":\"UPLOAD\"") == NULL || + strstr(fake.sent, "Y2FtZXJhIHt9Cg==") == NULL) { + fprintf(stderr, "upload request missing expected payload\n"); + return 6; + } + + fake.reply = "{\"type\":\"FAILED\",\"error\":\"staging failed\"}\n"; + if (amirender_upload_asset( + &transport, "scene.pov", scene, sizeof(scene) - 1, asset, sizeof(asset)) == 0) { + return 7; + } + + fake.reply = "{\"type\":\"DATA\",\"asset\":\"/tmp/amirender-asset-x/frame.png\"," + "\"data\":\"iVBORw0K\"}\n"; + if (amirender_download_asset( + &transport, "/tmp/amirender-asset-x/frame.png", + downloaded, sizeof(downloaded), &downloaded_size) != 0) { + return 8; + } + if (downloaded_size != 6 || + downloaded[0] != 0x89 || downloaded[1] != 'P' || + downloaded[2] != 'N' || downloaded[3] != 'G' || + downloaded[4] != 0x0d || downloaded[5] != 0x0a) { + fprintf(stderr, "downloaded data mismatch\n"); + return 9; + } + if (strstr(fake.sent, "\"type\":\"DOWNLOAD\"") == NULL) { + fprintf(stderr, "download request missing\n"); + return 10; + } + + fake.reply = "{\"type\":\"FAILED\",\"error\":\"asset unavailable\"}\n"; + if (amirender_download_asset( + &transport, "/tmp/amirender-asset-x/frame.png", + downloaded, sizeof(downloaded), &downloaded_size) == 0) { + return 11; + } + fake.reply = "{\"type\":\"COMPLETE\",\"output\":\"/tmp/old.png\",\"asset_id\":\"asset-new\"}"; + if (amirender_extract_output_asset(fake.reply, asset, sizeof(asset)) != 0 || + strcmp(asset, "asset-new") != 0) { + fprintf(stderr, "opaque asset ID not preferred\n"); + return 12; + } + fake.reply = "{\"type\":\"COMPLETE\",\"output\":\"/tmp/old.png\"}"; + if (amirender_extract_output_asset(fake.reply, asset, sizeof(asset)) != 0 || + strcmp(asset, "/tmp/old.png") != 0) { + fprintf(stderr, "legacy output fallback failed\n"); + return 13; + } + fake.reply = "{\"type\":\"COMPLETE\",\"output\":\"/tmp/old.png\",\"asset_id\":\"\"}"; + if (amirender_extract_output_asset(fake.reply, asset, sizeof(asset)) != 0 || + strcmp(asset, "/tmp/old.png") != 0) { + fprintf(stderr, "empty asset ID fallback failed\n"); + return 14; + } + fake.reply = "{\"type\":\"STAGING\",\"id\":\"asset-opaque\",\"asset\":\"/tmp/scene.pov\"}"; + if (amirender_upload_begin(&transport, "scene.pov", 10, asset, sizeof(asset)) != 0 || + strcmp(asset, "asset-opaque") != 0) { + fprintf(stderr, "opaque upload ID not preferred\n"); + return 15; + } + fake.reply = "{\"type\":\"STAGING\",\"asset\":\"/tmp/scene.pov\"}"; + if (amirender_upload_begin(&transport, "scene.pov", 10, asset, sizeof(asset)) != 0 || + strcmp(asset, "/tmp/scene.pov") != 0) { + fprintf(stderr, "legacy upload asset fallback failed\n"); + return 16; + } + fake.reply = "{\"type\":\"STAGING\",\"id\":\"asset-roundtrip\",\"asset\":\"/tmp/scene.pov\"}"; + if (amirender_upload_begin(&transport, "scene.pov", 10, asset, sizeof(asset)) != 0 || + strcmp(asset, "asset-roundtrip") != 0) return 17; + fake.reply = "{\"type\":\"CHUNKED\",\"offset\":10}"; + if (amirender_upload_chunk(&transport, asset, 0, scene, sizeof(scene) - 1) != 0 || + strstr(fake.sent, "\"asset\":\"asset-roundtrip\"") == NULL) { + fprintf(stderr, "upload chunk did not use opaque ID\n"); + return 18; + } + fake.reply = "{\"type\":\"STAGED\"}"; + if (amirender_upload_end(&transport, asset, 10) != 0 || + strstr(fake.sent, "\"asset\":\"asset-roundtrip\"") == NULL) { + fprintf(stderr, "upload end did not use opaque ID\n"); + return 19; + } + { + int eof = 0; + size_t count = 0; + fake.reply = "{\"type\":\"DATA\",\"data\":\"iVBORw0K\",\"eof\":false}"; + if (amirender_download_chunk(&transport, "asset-rendered", 0, + downloaded, sizeof(downloaded), &count, &eof) != 0 || + count != 6 || eof != 0 || + strstr(fake.sent, "\"asset\":\"asset-rendered\"") == NULL || + strstr(fake.sent, "\"offset\":0") == NULL || + downloaded[0] != 0x89 || downloaded[1] != 'P') { + fprintf(stderr, "first opaque download chunk failed\n"); + return 20; + } + fake.reply = "{\"type\":\"DATA\",\"data\":\"Tkc=\",\"eof\":true}"; + if (amirender_download_chunk(&transport, "asset-rendered", 6, + downloaded, sizeof(downloaded), &count, &eof) != 0 || + count != 2 || eof != 1 || + strstr(fake.sent, "\"asset\":\"asset-rendered\"") == NULL || + strstr(fake.sent, "\"offset\":6") == NULL || + downloaded[0] != 'N' || downloaded[1] != 'G') { + fprintf(stderr, "final opaque download chunk failed\n"); + return 21; + } + fake.reply = "{\"type\":\"DATA\",\"note\":\"escaped \\\"eof\\\":false and \\\"type\\\":\\\"FAILED\\\"\",\"data\":\"Tkc=\",\"eof\":true}"; + if (amirender_download_chunk(&transport, "asset-rendered", 6, + downloaded, sizeof(downloaded), &count, &eof) != 0 || + count != 2 || eof != 1 || + downloaded[0] != 'N' || downloaded[1] != 'G') { + fprintf(stderr, "escaped metadata must not shadow protocol fields\n"); + return 44; + } + fake.reply = "{\"type\":\"DATA\",\"eof\":true}"; + if (amirender_download_chunk(&transport, "asset-rendered", 8, + downloaded, sizeof(downloaded), &count, &eof) != 0 || + count != 0 || eof != 1) { + fprintf(stderr, "empty EOF response failed\n"); + return 22; + } + } + { + const char *path = "amirender-download-test.bin"; + const char *partial = "amirender-download-test.bin.part"; + FILE *file = fopen(path, "wb"); + size_t total = 999; + char existing[16] = {0}; + if (file == NULL) return 23; + if (fwrite("original", 1, 8, file) != 8 || fclose(file) != 0) return 24; + fake.reply = "{\"type\":\"FAILED\",\"error\":\"interrupted\"}"; + if (amirender_download_file(&transport, "asset-interrupted", path, &total) == 0) return 25; + file = fopen(path, "rb"); + if (file == NULL) return 26; + if (fread(existing, 1, 8, file) != 8 || fclose(file) != 0 || + memcmp(existing, "original", 8) != 0 || total != 999) return 27; + file = fopen(partial, "rb"); + if (file != NULL) { + fclose(file); + fprintf(stderr, "partial download was not removed\n"); + return 28; + } + remove(path); + } + { + const char *path = "amirender-midstream-test.bin"; + const char *partial = "amirender-midstream-test.bin.part"; + FILE *file = fopen(path, "wb"); + size_t total = 777; + char existing[16] = {0}; + if (file == NULL) return 29; + if (fwrite("original", 1, 8, file) != 8 || fclose(file) != 0) return 30; + fake.reply = "{\"type\":\"DATA\",\"data\":\"iVBORw0K\",\"eof\":false}"; + fake.next_reply = "{\"type\":\"FAILED\",\"error\":\"connection lost\"}"; + fake.receive_count = 0; + if (amirender_download_file(&transport, "asset-midstream", path, &total) == 0 || + fake.receive_count != 2 || + strstr(fake.sent, "\"offset\":6") == NULL) { + fprintf(stderr, "midstream failure was not detected\n"); + return 31; + } + fake.next_reply = NULL; + file = fopen(path, "rb"); + if (file == NULL) return 32; + if (fread(existing, 1, 8, file) != 8 || fclose(file) != 0 || + memcmp(existing, "original", 8) != 0 || total != 777) return 33; + file = fopen(partial, "rb"); + if (file != NULL) { + fclose(file); + fprintf(stderr, "midstream partial file was not removed\n"); + return 34; + } + remove(path); + } + { + const char *path = "amirender-complete-test.bin"; + const char *partial = "amirender-complete-test.bin.part"; + FILE *file = fopen(path, "wb"); + size_t total = 0; + unsigned char result[8] = {0}; + if (file == NULL) return 35; + if (fwrite("original", 1, 8, file) != 8 || fclose(file) != 0) return 36; + fake.reply = "{\"type\":\"DATA\",\"data\":\"iVBORw0K\",\"eof\":false}"; + fake.next_reply = "{\"type\":\"DATA\",\"data\":\"Tkc=\",\"eof\":true}"; + fake.receive_count = 0; + if (amirender_download_file(&transport, "asset-complete", path, &total) != 0 || + total != 8 || fake.receive_count != 2 || + strstr(fake.sent, "\"asset\":\"asset-complete\"") == NULL || + strstr(fake.sent, "\"offset\":6") == NULL) { + fprintf(stderr, "multi-chunk download did not complete\n"); + return 37; + } + fake.next_reply = NULL; + file = fopen(path, "rb"); + if (file == NULL) return 38; + if (fread(result, 1, sizeof(result), file) != sizeof(result) || + fclose(file) != 0 || + result[0] != 0x89 || result[1] != 'P' || result[2] != 'N' || result[3] != 'G' || + result[4] != 0x0d || result[5] != 0x0a || result[6] != 'N' || result[7] != 'G') { + fprintf(stderr, "published file content mismatch\n"); + return 39; + } + file = fopen(partial, "rb"); + if (file != NULL) { + fclose(file); + fprintf(stderr, "partial file remains after success\n"); + return 40; + } + remove(path); + } + { + const char *path = "amirender-invalid-test.bin"; + const char *partial = "amirender-invalid-test.bin.part"; + const char *invalid[] = { + "{\"type\":\"DATA\",\"data\":\"%%%?\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"QQ=A\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"QQ==QQ==\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"QR==\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"QUJ=\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"\",\"eof\":false}", + "{\"type\":\"DATA\",\"data\":\"\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"QQ==\"}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":null}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":1}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true,\"eof\":false}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":false,\"eof\":false}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":null,\"eof\":true}", + "{\"type\":\"DATA\",\"note\":\"\\\"eof\\\":true\",\"data\":\"QQ==\"}", + "{\"type\":\"FAILED\",\"note\":\"\\\"type\\\":\\\"DATA\\\"\",\"data\":\"QQ==\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"data\":\"QQ==\",\"eof\":true}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true,\"note\":\"unterminated}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true,\"note\":\"bad \\q escape\"}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true,\"note\":\"bad \\u12G4 escape\"}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true,\"note\":\"bad \\u123 escape\"}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true}}", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true}junk", + "{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true}{\"x\":1}", + "garbage{\"type\":\"DATA\",\"data\":\"QQ==\",\"eof\":true}", + "{\"type\":\"DATA\",\"eof\":true}", + "{\"type\":\"DATA\",\"eof\":false}", + "{\"type\":\"FAILED\",\"error\":\"disconnected\"}" + }; + size_t case_index; + for (case_index = 0; case_index < sizeof(invalid) / sizeof(invalid[0]); ++case_index) { + FILE *file = fopen(path, "wb"); + char existing[8] = {0}; + size_t total = 123; + if (file == NULL) return 41; + if (fwrite("original", 1, 8, file) != 8 || fclose(file) != 0) return 42; + fake.reply = invalid[case_index]; + fake.next_reply = NULL; + fake.receive_count = 0; + if (amirender_download_file(&transport, "asset-invalid", path, &total) == 0 || + total != 123) { + fprintf(stderr, "invalid download case %lu unexpectedly succeeded: %s\n", + (unsigned long)case_index, invalid[case_index]); + return 43; + } + file = fopen(path, "rb"); + if (file == NULL) return 44; + if (fread(existing, 1, 8, file) != 8 || fclose(file) != 0 || + memcmp(existing, "original", 8) != 0) return 45; + file = fopen(partial, "rb"); + if (file != NULL) { + fclose(file); + return 46; + } + } + remove(path); + } + return 0; +} diff --git a/clients/amiga/tests/test_submit.c b/clients/amiga/tests/test_submit.c new file mode 100644 index 0000000..b5e2e74 --- /dev/null +++ b/clients/amiga/tests/test_submit.c @@ -0,0 +1,30 @@ +#include "amirender_submit.h" + +#include +#include + +int main(void) +{ + char buffer[512]; + struct amirender_job job = { + "amiga-0001", "povray", "DH0:Scenes/demo.pov", + "RAM:frame.png", 0, 320, 256 + }; + const char *expected = + "{\"type\":\"SUBMIT\",\"job\":{\"id\":\"amiga-0001\",\"engine\":\"povray\"," + "\"scene\":\"DH0:Scenes/demo.pov\",\"output\":\"RAM:frame.png\"," + "\"frame\":0,\"width\":320,\"height\":256}}\n"; + + if (amirender_build_submit(buffer, sizeof(buffer), &job) < 0) { + return 1; + } + if (strcmp(buffer, expected) != 0) { + fprintf(stderr, "unexpected message: %s", buffer); + return 2; + } + job.scene = "bad\"scene.pov"; + if (amirender_build_submit(buffer, sizeof(buffer), &job) >= 0) { + return 3; + } + return 0; +} diff --git a/clients/amiga/tests/test_upload.c b/clients/amiga/tests/test_upload.c new file mode 100644 index 0000000..ea9eefc --- /dev/null +++ b/clients/amiga/tests/test_upload.c @@ -0,0 +1,21 @@ +#include "amirender_upload.h" + +#include +#include + +int main(void) +{ + char buffer[256]; + const unsigned char scene[] = "camera {}\n"; + const char *expected = + "{\"type\":\"UPLOAD\",\"name\":\"scene.pov\",\"data\":\"Y2FtZXJhIHt9Cg==\"}\n"; + + if (amirender_build_upload(buffer, sizeof(buffer), "scene.pov", scene, sizeof(scene) - 1) < 0) return 1; + if (strcmp(buffer, expected) != 0) { + fprintf(stderr, "unexpected upload: %s", buffer); + return 2; + } + if (amirender_build_upload(buffer, sizeof(buffer), "../scene.pov", scene, sizeof(scene) - 1) >= 0) return 3; + if (amirender_build_upload(buffer, 16, "scene.pov", scene, sizeof(scene) - 1) >= 0) return 4; + return 0; +} diff --git a/cmd/amirender-node/main.go b/cmd/amirender-node/main.go index cc75124..957cb8b 100644 --- a/cmd/amirender-node/main.go +++ b/cmd/amirender-node/main.go @@ -1,42 +1,12 @@ package main import ( - "bufio" - "encoding/json" "fmt" "net" "os" -) - -type Message struct { - Type string `json:"type"` - Job json.RawMessage `json:"job"` -} - -type Job struct { - ID string `json:"id"` - Engine string `json:"engine"` - Output string `json:"output"` -} -func handle(c net.Conn) { - defer c.Close() - s := bufio.NewScanner(c) - for s.Scan() { - var m Message - if json.Unmarshal(s.Bytes(), &m) != nil || m.Type != "SUBMIT" { - fmt.Fprintln(c, `{"type":"FAILED","error":"bad request"}`) - continue - } - var j Job - if json.Unmarshal(m.Job, &j) != nil || j.Engine != "null" { - fmt.Fprintln(c, `{"type":"FAILED","error":"unsupported engine"}`) - continue - } - out, _ := json.Marshal(map[string]any{"type": "COMPLETE", "job_id": j.ID, "engine": "null", "output": j.Output}) - fmt.Fprintln(c, string(out)) - } -} + "github.com/Ploos-AS/AmiRender/internal/node" +) func main() { addr := "127.0.0.1:6800" @@ -48,12 +18,8 @@ func main() { panic(err) } defer ln.Close() - fmt.Printf("AmiRender M0 node listening on %s\n", addr) - for { - c, err := ln.Accept() - if err != nil { - panic(err) - } - go handle(c) + fmt.Printf("AmiRender node listening on %s\n", addr) + if err := node.Serve(ln); err != nil { + panic(err) } } diff --git a/cmd/amirender-node/main_test.go b/cmd/amirender-node/main_test.go index 5596e83..e0b1178 100644 --- a/cmd/amirender-node/main_test.go +++ b/cmd/amirender-node/main_test.go @@ -6,6 +6,8 @@ import ( "net" "testing" "time" + + "github.com/Ploos-AS/AmiRender/internal/node" ) func TestNullEngineEndToEnd(t *testing.T) { @@ -18,7 +20,7 @@ func TestNullEngineEndToEnd(t *testing.T) { go func() { c, err := ln.Accept() if err == nil { - handle(c) + node.Handle(c) } }() diff --git a/config/idle.example.json b/config/idle.example.json new file mode 100644 index 0000000..b0e7937 --- /dev/null +++ b/config/idle.example.json @@ -0,0 +1,17 @@ +{ + "enabled": false, + "workload": "none", + "idle_delay_seconds": 300, + "cpu_percent": 50, + "temperature_limit_c": 75, + "monero": { + "executable": "", + "pool": "", + "wallet": "" + }, + "verus": { + "executable": "", + "pool": "", + "wallet": "" + } +} diff --git a/docs/FARM.md b/docs/FARM.md new file mode 100644 index 0000000..12a8808 --- /dev/null +++ b/docs/FARM.md @@ -0,0 +1,28 @@ +# AmiRender farm model + +AmiRender uses one engine-agnostic farm for all render backends. + +A node advertises capabilities such as: + +- `povray` +- `blender-cycles` +- `blender-eevee` +- future `ami3d`, `amiterrain` and translated classic engines + +The controller schedules a job only to a node advertising the requested engine. + +## POV-Ray + +POV-Ray is a first-class farm engine, not a local-only appliance backend. + +M1 targets frame-level distribution. Each animation frame is an independent render job and may execute on any capable ARM64 or AMD64 node. A single AmiRender appliance may act as both controller and worker. + +Region/tile distribution for large still images is intentionally deferred until the frame scheduler is stable. + +## Blender + +Blender follows the same farm contract. CPU/GPU-specific capabilities may be advertised separately so the scheduler can prefer appropriate workers without exposing those implementation details to Amiga clients. + +## Scheduler evolution + +M1 uses deterministic first-capable-node selection as a testable baseline. Later milestones may add measured performance weights, current load, asset locality, architecture, memory and GPU capability. diff --git a/docs/IDLE-WORKLOADS.md b/docs/IDLE-WORKLOADS.md new file mode 100644 index 0000000..32f2f3c --- /dev/null +++ b/docs/IDLE-WORKLOADS.md @@ -0,0 +1,39 @@ +# Optional idle workloads + +AmiRender may use otherwise idle ARM64/AMD64 worker capacity for explicitly enabled background workloads. + +## Policy + +- disabled by default +- explicit opt-in per node +- rendering always has absolute priority +- idle work is stopped before render dispatch +- configurable idle delay and resource limits +- credentials, wallet addresses and pool configuration are supplied by the node owner; AmiRender ships none +- idle and render accounting remain separate +- m68k workers do not enable idle mining by default + +## Initial adapters + +Planned initial adapters: + +- Monero / RandomX +- Verus / VerusHash + +Mining software is an external dependency. AmiRender does not vendor miners and does not contain default wallets or mining-pool credentials. + +## Generality + +The subsystem is intentionally named idle workloads rather than mining. Future adapters may include CI, transcoding or other owner-authorized batch work. + +## Node lifecycle + +IDLE -> IDLE_DELAY -> IDLE_WORKLOAD + +When a render job arrives: + +IDLE_WORKLOAD -> PREEMPTING -> RENDERING + +After the render queue drains: + +RENDERING -> IDLE -> IDLE_DELAY -> IDLE_WORKLOAD diff --git a/docs/M1.md b/docs/M1.md new file mode 100644 index 0000000..1600aa2 --- /dev/null +++ b/docs/M1.md @@ -0,0 +1,16 @@ +# M1 — POV-Ray backend and first real client path + +## Goals + +1. Promote the M0 engine contract into Go packages. +2. Add POV-Ray as the first production renderer. +3. Execute POV-Ray without a shell. +4. Add a deterministic smoke scene. +5. Integrate the adapter with the node protocol. +6. Add the first AmigaOS/m68k submission client. + +POV-Ray is an external runtime dependency. AmiRender does not vendor or redistribute it. + +## Security boundary + +Scene paths and render options are passed as process arguments, never interpolated into a shell command. The appliance will later add job-root path confinement and an asset cache. diff --git a/internal/engine/engine.go b/internal/engine/engine.go new file mode 100644 index 0000000..1f302f3 --- /dev/null +++ b/internal/engine/engine.go @@ -0,0 +1,23 @@ +package engine + +import "context" + +type Job struct { + ID string + Scene string + Output string + Frame int + Width int + Height int + Options map[string]string +} + +type Result struct { + Output string +} + +type Engine interface { + Name() string + Probe(context.Context) error + Render(context.Context, Job) (Result, error) +} diff --git a/internal/engine/povray/povray.go b/internal/engine/povray/povray.go new file mode 100644 index 0000000..caa4769 --- /dev/null +++ b/internal/engine/povray/povray.go @@ -0,0 +1,51 @@ +package povray + +import ( + "context" + "fmt" + "os/exec" + "strconv" + + "github.com/Ploos-AS/AmiRender/internal/engine" +) + +type Engine struct { + Binary string +} + +func New(binary string) *Engine { + if binary == "" { + binary = "povray" + } + return &Engine{Binary: binary} +} + +func (e *Engine) Name() string { return "povray" } + +func (e *Engine) Probe(ctx context.Context) error { + _, err := exec.LookPath(e.Binary) + if err != nil { + return fmt.Errorf("povray executable %q not found: %w", e.Binary, err) + } + return nil +} + +func (e *Engine) Render(ctx context.Context, job engine.Job) (engine.Result, error) { + if job.Scene == "" || job.Output == "" { + return engine.Result{}, fmt.Errorf("scene and output are required") + } + + args := []string{"+I" + job.Scene, "+O" + job.Output, "-D"} + if job.Width > 0 { + args = append(args, "+W"+strconv.Itoa(job.Width)) + } + if job.Height > 0 { + args = append(args, "+H"+strconv.Itoa(job.Height)) + } + + cmd := exec.CommandContext(ctx, e.Binary, args...) + if out, err := cmd.CombinedOutput(); err != nil { + return engine.Result{}, fmt.Errorf("povray render failed: %w: %s", err, string(out)) + } + return engine.Result{Output: job.Output}, nil +} diff --git a/internal/engine/povray/povray_test.go b/internal/engine/povray/povray_test.go new file mode 100644 index 0000000..36564c7 --- /dev/null +++ b/internal/engine/povray/povray_test.go @@ -0,0 +1,44 @@ +package povray + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/Ploos-AS/AmiRender/internal/engine" +) + +func TestRenderUsesArgumentVectorWithoutShell(t *testing.T) { + dir := t.TempDir() + log := filepath.Join(dir, "args.txt") + fake := filepath.Join(dir, "povray") + script := "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"" + log + "\"\n" + if err := os.WriteFile(fake, []byte(script), 0755); err != nil { + t.Fatal(err) + } + + e := New(fake) + scene := "scene;touch SHOULD_NOT_EXIST.pov" + _, err := e.Render(context.Background(), engine.Job{ + ID: "m1-test", Scene: scene, Output: "frame.png", Width: 320, Height: 256, + }) + if err != nil { + t.Fatal(err) + } + + data, err := os.ReadFile(log) + if err != nil { + t.Fatal(err) + } + got := string(data) + for _, want := range []string{"+I" + scene, "+Oframe.png", "-D", "+W320", "+H256"} { + if !strings.Contains(got, want) { + t.Fatalf("missing argument %q in %q", want, got) + } + } + if _, err := os.Stat("SHOULD_NOT_EXIST.pov"); !os.IsNotExist(err) { + t.Fatal("scene name was interpreted by a shell") + } +} diff --git a/internal/farm/capability.go b/internal/farm/capability.go new file mode 100644 index 0000000..3844577 --- /dev/null +++ b/internal/farm/capability.go @@ -0,0 +1,21 @@ +package farm + +type Capability struct { + Engine string `json:"engine"` + Modes []string `json:"modes,omitempty"` +} + +type Node struct { + ID string `json:"id"` + Arch string `json:"arch"` + Capabilities []Capability `json:"capabilities"` +} + +func (n Node) Supports(engine string) bool { + for _, c := range n.Capabilities { + if c.Engine == engine { + return true + } + } + return false +} diff --git a/internal/farm/coordinator.go b/internal/farm/coordinator.go new file mode 100644 index 0000000..aca4f65 --- /dev/null +++ b/internal/farm/coordinator.go @@ -0,0 +1,32 @@ +package farm + +import ( + "context" + "fmt" +) + +type EndpointNode struct { + Node + Address string +} + +type Coordinator struct { + Registry *Registry + Endpoints map[string]string + Client WorkerClient +} + +func (c Coordinator) Render(ctx context.Context, job RenderJob) (WorkerResult, error) { + if c.Registry == nil || c.Client == nil { + return WorkerResult{}, fmt.Errorf("registry and worker client are required") + } + assignment, err := (Dispatcher{Registry: c.Registry}).Assign(job.ID, job.Engine) + if err != nil { + return WorkerResult{}, err + } + address := c.Endpoints[assignment.Node.ID] + if address == "" { + return WorkerResult{}, fmt.Errorf("node %q has no endpoint", assignment.Node.ID) + } + return c.Client.Render(ctx, address, job) +} diff --git a/internal/farm/coordinator_test.go b/internal/farm/coordinator_test.go new file mode 100644 index 0000000..fb9540b --- /dev/null +++ b/internal/farm/coordinator_test.go @@ -0,0 +1,81 @@ +package farm + +import ( + "context" + "encoding/json" + "net" + "testing" + "time" +) + +func TestCoordinatorRoutesJobToCapableWorker(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + received := make(chan RenderJob, 1) + go func() { + c, err := ln.Accept() + if err != nil { + return + } + defer c.Close() + var msg struct { + Type string `json:"type"` + Job RenderJob `json:"job"` + } + if json.NewDecoder(c).Decode(&msg) != nil { + return + } + received <- msg.Job + _ = json.NewEncoder(c).Encode(WorkerResult{ + Type: "COMPLETE", JobID: msg.Job.ID, Engine: msg.Job.Engine, Output: msg.Job.Output, + }) + }() + + r := NewRegistry() + if err := r.Register(Node{ + ID: "rock5-01", Arch: "arm64", + Capabilities: []Capability{{Engine: "povray"}}, + }); err != nil { + t.Fatal(err) + } + if err := r.Register(Node{ + ID: "gpu-01", Arch: "amd64", + Capabilities: []Capability{{Engine: "blender-cycles"}}, + }); err != nil { + t.Fatal(err) + } + + coordinator := Coordinator{ + Registry: r, + Endpoints: map[string]string{ + "rock5-01": ln.Addr().String(), + }, + Client: TCPWorkerClient{}, + } + job := RenderJob{ + ID: "frame-0042", Engine: "povray", Scene: "m1-smoke.pov", + Output: "frame0042.png", Frame: 42, Width: 160, Height: 120, + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + result, err := coordinator.Render(ctx, job) + if err != nil { + t.Fatal(err) + } + if result.Type != "COMPLETE" || result.JobID != job.ID { + t.Fatalf("unexpected result: %#v", result) + } + + select { + case got := <-received: + if got.Engine != "povray" || got.ID != job.ID { + t.Fatalf("wrong job reached worker: %#v", got) + } + case <-ctx.Done(): + t.Fatal("worker did not receive job") + } +} diff --git a/internal/farm/dispatch.go b/internal/farm/dispatch.go new file mode 100644 index 0000000..31d4624 --- /dev/null +++ b/internal/farm/dispatch.go @@ -0,0 +1,28 @@ +package farm + +import "fmt" + +type Assignment struct { + JobID string + Engine string + Node Node +} + +type Dispatcher struct { + Registry *Registry +} + +func (d Dispatcher) Assign(jobID, engine string) (Assignment, error) { + if d.Registry == nil { + return Assignment{}, fmt.Errorf("registry is required") + } + if jobID == "" || engine == "" { + return Assignment{}, fmt.Errorf("job id and engine are required") + } + candidates := d.Registry.NodesFor(engine) + node, err := SelectNode(candidates, engine) + if err != nil { + return Assignment{}, err + } + return Assignment{JobID: jobID, Engine: engine, Node: node}, nil +} diff --git a/internal/farm/dispatch_test.go b/internal/farm/dispatch_test.go new file mode 100644 index 0000000..1282f79 --- /dev/null +++ b/internal/farm/dispatch_test.go @@ -0,0 +1,44 @@ +package farm + +import ( + "errors" + "testing" +) + +func TestDispatcherRoutesPOVRayToRegisteredWorker(t *testing.T) { + r := NewRegistry() + if err := r.Register(Node{ + ID: "rock5-01", Arch: "arm64", + Capabilities: []Capability{{Engine: "povray"}}, + }); err != nil { + t.Fatal(err) + } + if err := r.Register(Node{ + ID: "gpu-01", Arch: "amd64", + Capabilities: []Capability{{Engine: "blender-cycles"}}, + }); err != nil { + t.Fatal(err) + } + + a, err := (Dispatcher{Registry: r}).Assign("frame-0042", "povray") + if err != nil { + t.Fatal(err) + } + if a.Node.ID != "rock5-01" || a.Engine != "povray" || a.JobID != "frame-0042" { + t.Fatalf("unexpected assignment: %#v", a) + } +} + +func TestDispatcherRejectsMissingCapability(t *testing.T) { + r := NewRegistry() + if err := r.Register(Node{ + ID: "rock5-01", Arch: "arm64", + Capabilities: []Capability{{Engine: "povray"}}, + }); err != nil { + t.Fatal(err) + } + _, err := (Dispatcher{Registry: r}).Assign("frame-0001", "blender-cycles") + if !errors.Is(err, ErrNoCapableNode) { + t.Fatalf("want ErrNoCapableNode, got %v", err) + } +} diff --git a/internal/farm/registry.go b/internal/farm/registry.go new file mode 100644 index 0000000..76492de --- /dev/null +++ b/internal/farm/registry.go @@ -0,0 +1,48 @@ +package farm + +import ( + "fmt" + "sort" + "sync" +) + +type Registry struct { + mu sync.RWMutex + nodes map[string]Node +} + +func NewRegistry() *Registry { + return &Registry{nodes: make(map[string]Node)} +} + +func (r *Registry) Register(node Node) error { + if node.ID == "" { + return fmt.Errorf("node id is required") + } + if len(node.Capabilities) == 0 { + return fmt.Errorf("node %q has no capabilities", node.ID) + } + r.mu.Lock() + defer r.mu.Unlock() + r.nodes[node.ID] = node + return nil +} + +func (r *Registry) Remove(id string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.nodes, id) +} + +func (r *Registry) NodesFor(engine string) []Node { + r.mu.RLock() + defer r.mu.RUnlock() + var out []Node + for _, node := range r.nodes { + if node.Supports(engine) { + out = append(out, node) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out +} diff --git a/internal/farm/registry_test.go b/internal/farm/registry_test.go new file mode 100644 index 0000000..a68e5fa --- /dev/null +++ b/internal/farm/registry_test.go @@ -0,0 +1,36 @@ +package farm + +import "testing" + +func TestRegistryFiltersByCapability(t *testing.T) { + r := NewRegistry() + for _, n := range []Node{ + {ID: "rock5-01", Arch: "arm64", Capabilities: []Capability{{Engine: "povray"}}}, + {ID: "gpu-01", Arch: "amd64", Capabilities: []Capability{{Engine: "povray"}, {Engine: "blender-cycles"}}}, + } { + if err := r.Register(n); err != nil { + t.Fatal(err) + } + } + pov := r.NodesFor("povray") + if len(pov) != 2 { + t.Fatalf("want 2 POV-Ray nodes, got %d", len(pov)) + } + blender := r.NodesFor("blender-cycles") + if len(blender) != 1 || blender[0].ID != "gpu-01" { + t.Fatalf("unexpected Blender nodes: %#v", blender) + } +} + +func TestReregistrationUpdatesCapabilities(t *testing.T) { + r := NewRegistry() + if err := r.Register(Node{ID: "node-01", Arch: "arm64", Capabilities: []Capability{{Engine: "povray"}}}); err != nil { + t.Fatal(err) + } + if err := r.Register(Node{ID: "node-01", Arch: "arm64", Capabilities: []Capability{{Engine: "blender-cycles"}}}); err != nil { + t.Fatal(err) + } + if len(r.NodesFor("povray")) != 0 || len(r.NodesFor("blender-cycles")) != 1 { + t.Fatal("re-registration did not replace capabilities") + } +} diff --git a/internal/farm/scheduler.go b/internal/farm/scheduler.go new file mode 100644 index 0000000..e9d1452 --- /dev/null +++ b/internal/farm/scheduler.go @@ -0,0 +1,17 @@ +package farm + +import "errors" + +var ErrNoCapableNode = errors.New("no capable node") + +// SelectNode is deliberately simple in M1: first capable node wins. +// Later schedulers can add load, benchmark weights, locality and GPU preference +// without changing the job or engine contracts. +func SelectNode(nodes []Node, engine string) (Node, error) { + for _, n := range nodes { + if n.Supports(engine) { + return n, nil + } + } + return Node{}, ErrNoCapableNode +} diff --git a/internal/farm/scheduler_test.go b/internal/farm/scheduler_test.go new file mode 100644 index 0000000..435a0e9 --- /dev/null +++ b/internal/farm/scheduler_test.go @@ -0,0 +1,18 @@ +package farm + +import "testing" + +func TestSelectNodeByEngineCapability(t *testing.T) { + nodes := []Node{ + {ID: "arm-01", Arch: "arm64", Capabilities: []Capability{{Engine: "povray"}}}, + {ID: "gpu-01", Arch: "amd64", Capabilities: []Capability{{Engine: "povray"}, {Engine: "blender-cycles"}}}, + } + n, err := SelectNode(nodes, "povray") + if err != nil || n.ID != "arm-01" { + t.Fatalf("unexpected selection: %#v %v", n, err) + } + n, err = SelectNode(nodes, "blender-cycles") + if err != nil || n.ID != "gpu-01" { + t.Fatalf("unexpected Blender selection: %#v %v", n, err) + } +} diff --git a/internal/farm/transport.go b/internal/farm/transport.go new file mode 100644 index 0000000..ca34552 --- /dev/null +++ b/internal/farm/transport.go @@ -0,0 +1,63 @@ +package farm + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "net" +) + +type RenderJob struct { + ID string `json:"id"` + Engine string `json:"engine"` + Scene string `json:"scene"` + Output string `json:"output"` + Frame int `json:"frame,omitempty"` + Width int `json:"width,omitempty"` + Height int `json:"height,omitempty"` +} + +type WorkerResult struct { + Type string `json:"type"` + JobID string `json:"job_id"` + Engine string `json:"engine,omitempty"` + Output string `json:"output,omitempty"` + AssetID string `json:"asset_id,omitempty"` + Error string `json:"error,omitempty"` +} + +type WorkerClient interface { + Render(context.Context, string, RenderJob) (WorkerResult, error) +} + +type TCPWorkerClient struct{} + +func (TCPWorkerClient) Render(ctx context.Context, address string, job RenderJob) (WorkerResult, error) { + var result WorkerResult + d := net.Dialer{} + conn, err := d.DialContext(ctx, "tcp", address) + if err != nil { + return result, err + } + defer conn.Close() + + msg := struct { + Type string `json:"type"` + Job RenderJob `json:"job"` + }{Type: "SUBMIT", Job: job} + if err := json.NewEncoder(conn).Encode(msg); err != nil { + return result, err + } + if err := json.NewDecoder(bufio.NewReader(conn)).Decode(&result); err != nil { + return result, err + } + switch result.Type { + case "COMPLETE": + return result, nil + case "FAILED": + return result, fmt.Errorf("worker failed: %s", result.Error) + default: + return result, fmt.Errorf("unexpected worker response %q", result.Type) + } +} diff --git a/internal/farm/transport_test.go b/internal/farm/transport_test.go new file mode 100644 index 0000000..be244b0 --- /dev/null +++ b/internal/farm/transport_test.go @@ -0,0 +1,47 @@ +package farm + +import ( + "context" + "encoding/json" + "net" + "testing" + "time" +) + +func TestTCPWorkerClientRoundTrip(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + + go func() { + c, err := ln.Accept() + if err != nil { + return + } + defer c.Close() + var msg struct { + Type string `json:"type"` + Job RenderJob `json:"job"` + } + if json.NewDecoder(c).Decode(&msg) != nil { + return + } + _ = json.NewEncoder(c).Encode(WorkerResult{ + Type: "COMPLETE", JobID: msg.Job.ID, Engine: msg.Job.Engine, Output: msg.Job.Output, + }) + }() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + result, err := (TCPWorkerClient{}).Render(ctx, ln.Addr().String(), RenderJob{ + ID: "frame-0042", Engine: "povray", Scene: "scene.pov", Output: "frame0042.png", + }) + if err != nil { + t.Fatal(err) + } + if result.Type != "COMPLETE" || result.JobID != "frame-0042" || result.Engine != "povray" { + t.Fatalf("unexpected result: %#v", result) + } +} diff --git a/internal/idle/mining/config.go b/internal/idle/mining/config.go new file mode 100644 index 0000000..1b81386 --- /dev/null +++ b/internal/idle/mining/config.go @@ -0,0 +1,33 @@ +package mining + +import "fmt" + +// Config describes an owner-supplied external idle workload. +// AmiRender stores policy and configuration only; execution is delegated +// outside the render node. +type Config struct { + Enabled bool + Kind string + Executable string + Pool string + Wallet string + CPUPercent int +} + +func (c Config) Validate() error { + if !c.Enabled { + return nil + } + switch c.Kind { + case "monero", "verus": + default: + return fmt.Errorf("unsupported mining workload %q", c.Kind) + } + if c.Executable == "" || c.Pool == "" || c.Wallet == "" { + return fmt.Errorf("%s requires owner-supplied executable, pool and wallet", c.Kind) + } + if c.CPUPercent < 1 || c.CPUPercent > 100 { + return fmt.Errorf("cpu percent must be between 1 and 100") + } + return nil +} diff --git a/internal/idle/mining/config_test.go b/internal/idle/mining/config_test.go new file mode 100644 index 0000000..bd16d07 --- /dev/null +++ b/internal/idle/mining/config_test.go @@ -0,0 +1,30 @@ +package mining + +import "testing" + +func TestDisabledConfigurationNeedsNoCredentials(t *testing.T) { + if err := (Config{}).Validate(); err != nil { + t.Fatal(err) + } +} + +func TestEnabledProfilesRequireOwnerConfiguration(t *testing.T) { + for _, kind := range []string{"monero", "verus"} { + c := Config{Enabled: true, Kind: kind, CPUPercent: 50} + if err := c.Validate(); err == nil { + t.Fatalf("%s accepted missing owner configuration", kind) + } + } +} + +func TestConfiguredProfilesValidate(t *testing.T) { + for _, kind := range []string{"monero", "verus"} { + c := Config{ + Enabled: true, Kind: kind, Executable: "/external/worker", + Pool: "owner-configured", Wallet: "owner-configured", CPUPercent: 50, + } + if err := c.Validate(); err != nil { + t.Fatalf("%s: %v", kind, err) + } + } +} diff --git a/internal/idle/state.go b/internal/idle/state.go new file mode 100644 index 0000000..d75aada --- /dev/null +++ b/internal/idle/state.go @@ -0,0 +1,61 @@ +package idle + +import "fmt" + +type State string + +const ( + StateIdle State = "IDLE" + StateDelay State = "IDLE_DELAY" + StateWorkload State = "IDLE_WORKLOAD" + StatePreempting State = "PREEMPTING" + StateRendering State = "RENDERING" +) + +type Machine struct{ State State } + +func NewMachine() *Machine { return &Machine{State: StateIdle} } + +func (m *Machine) IdleTimerStarted() error { + if m.State != StateIdle { + return fmt.Errorf("cannot start idle delay from %s", m.State) + } + m.State = StateDelay + return nil +} + +func (m *Machine) IdleTimerExpired() error { + if m.State != StateDelay { + return fmt.Errorf("cannot start workload from %s", m.State) + } + m.State = StateWorkload + return nil +} + +func (m *Machine) RenderArrived() error { + switch m.State { + case StateIdle, StateDelay: + m.State = StateRendering + case StateWorkload: + m.State = StatePreempting + default: + return fmt.Errorf("cannot accept render from %s", m.State) + } + return nil +} + +func (m *Machine) Preempted() error { + if m.State != StatePreempting { + return fmt.Errorf("cannot finish preemption from %s", m.State) + } + m.State = StateRendering + return nil +} + +func (m *Machine) RenderQueueDrained() error { + if m.State != StateRendering { + return fmt.Errorf("cannot drain render queue from %s", m.State) + } + m.State = StateIdle + return nil +} diff --git a/internal/idle/state_test.go b/internal/idle/state_test.go new file mode 100644 index 0000000..78dfb60 --- /dev/null +++ b/internal/idle/state_test.go @@ -0,0 +1,38 @@ +package idle + +import "testing" + +func TestIdleWorkloadPreemptLifecycle(t *testing.T) { + m := NewMachine() + steps := []struct { + want State + fn func() error + }{ + {StateDelay, m.IdleTimerStarted}, + {StateWorkload, m.IdleTimerExpired}, + {StatePreempting, m.RenderArrived}, + {StateRendering, m.Preempted}, + {StateIdle, m.RenderQueueDrained}, + } + for _, s := range steps { + if err := s.fn(); err != nil { + t.Fatal(err) + } + if m.State != s.want { + t.Fatalf("want %s, got %s", s.want, m.State) + } + } +} + +func TestRenderSkipsIdleWorkloadDuringDelay(t *testing.T) { + m := NewMachine() + if err := m.IdleTimerStarted(); err != nil { + t.Fatal(err) + } + if err := m.RenderArrived(); err != nil { + t.Fatal(err) + } + if m.State != StateRendering { + t.Fatalf("want rendering, got %s", m.State) + } +} diff --git a/internal/idle/workload.go b/internal/idle/workload.go new file mode 100644 index 0000000..c849467 --- /dev/null +++ b/internal/idle/workload.go @@ -0,0 +1,42 @@ +package idle + +import ( + "context" + "errors" + "time" +) + +var ErrDisabled = errors.New("idle workload disabled") + +type Config struct { + Enabled bool + IdleDelay time.Duration + CPUPercent int +} + +type Workload interface { + Name() string + Start(context.Context) error + Stop() error +} + +type Manager struct { + Config Config + Workload Workload +} + +func (m *Manager) StartIfIdle(ctx context.Context) error { + if !m.Config.Enabled { + return ErrDisabled + } + return m.Workload.Start(ctx) +} + +// Preempt is called before render work is dispatched. +// Idle work must never block rendering. +func (m *Manager) Preempt() error { + if m.Workload == nil { + return nil + } + return m.Workload.Stop() +} diff --git a/internal/idle/workload_test.go b/internal/idle/workload_test.go new file mode 100644 index 0000000..0af33c3 --- /dev/null +++ b/internal/idle/workload_test.go @@ -0,0 +1,38 @@ +package idle + +import ( + "context" + "errors" + "testing" +) + +type fakeWorkload struct{ started, stopped bool } + +func (f *fakeWorkload) Name() string { return "fake" } +func (f *fakeWorkload) Start(context.Context) error { f.started = true; return nil } +func (f *fakeWorkload) Stop() error { f.stopped = true; return nil } + +func TestDisabledByDefault(t *testing.T) { + f := &fakeWorkload{} + m := Manager{Workload: f} + if !errors.Is(m.StartIfIdle(context.Background()), ErrDisabled) { + t.Fatal("idle workload must be disabled by default") + } + if f.started { + t.Fatal("disabled workload started") + } +} + +func TestRenderCanPreemptIdleWork(t *testing.T) { + f := &fakeWorkload{} + m := Manager{Config: Config{Enabled: true}, Workload: f} + if err := m.StartIfIdle(context.Background()); err != nil { + t.Fatal(err) + } + if err := m.Preempt(); err != nil { + t.Fatal(err) + } + if !f.stopped { + t.Fatal("idle workload was not stopped") + } +} diff --git a/internal/node/server.go b/internal/node/server.go new file mode 100644 index 0000000..ce66c72 --- /dev/null +++ b/internal/node/server.go @@ -0,0 +1,517 @@ +package node + +import ( + "bufio" + "context" + "crypto/rand" + "encoding/base64" + "encoding/hex" + "encoding/json" + "fmt" + "net" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/Ploos-AS/AmiRender/internal/engine" + "github.com/Ploos-AS/AmiRender/internal/engine/povray" + "github.com/Ploos-AS/AmiRender/internal/farm" +) + +type uploadSession struct { + size int64 + created time.Time + dir string +} + +const uploadSessionTTL = 30 * time.Minute +const completedAssetTTL = 24 * time.Hour + +var uploadSessions = struct { + sync.Mutex + expected map[string]uploadSession +}{expected: make(map[string]uploadSession)} + +// assetIDs is a transitional registry. Existing clients continue to use paths +// until the protocol and Amiga client can switch to opaque identifiers. +var assetIDs = struct { + sync.Mutex + paths map[string]string + completed map[string]time.Time + active map[string]int +}{paths: make(map[string]string), completed: make(map[string]time.Time), active: make(map[string]int)} + +func newAssetID(path string) (string, error) { + var random [16]byte + if _, err := rand.Read(random[:]); err != nil { + return "", err + } + id := "asset-" + hex.EncodeToString(random[:]) + assetIDs.Lock() + assetIDs.paths[id] = path + assetIDs.Unlock() + return id, nil +} + +// resolveUploadAsset accepts opaque IDs and legacy paths during migration. +func markAssetCompleted(path string, now time.Time) { + assetIDs.Lock() + for id, registeredPath := range assetIDs.paths { + if registeredPath == path { + assetIDs.completed[id] = now + } + } + assetIDs.Unlock() +} + +func resolveUploadAsset(asset string) string { + if !strings.HasPrefix(asset, "asset-") { + return asset + } + assetIDs.Lock() + path := assetIDs.paths[asset] + assetIDs.Unlock() + return path +} + +const maxUploadBytes = 1024 * 1024 +const maxMessageBytes = 2 * 1024 * 1024 + +type message struct { + Type string `json:"type"` + Job json.RawMessage `json:"job"` + Name string `json:"name,omitempty"` + Data string `json:"data,omitempty"` + Asset string `json:"asset,omitempty"` + Offset int64 `json:"offset,omitempty"` + Size int64 `json:"size,omitempty"` +} + +type stagedResult struct { + ID string `json:"id,omitempty"` + Type string `json:"type"` + Asset string `json:"asset,omitempty"` + Error string `json:"error,omitempty"` + Data string `json:"data,omitempty"` + Offset int64 `json:"offset,omitempty"` + Size int64 `json:"size,omitempty"` + EOF bool `json:"eof,omitempty"` +} + +func Handle(c net.Conn) { + defer c.Close() + s := bufio.NewScanner(c) + s.Buffer(make([]byte, 64*1024), maxMessageBytes) + for s.Scan() { + var m message + if json.Unmarshal(s.Bytes(), &m) != nil { + writeResult(c, farm.WorkerResult{Type: "FAILED", Error: "bad request"}) + continue + } + if m.Type == "UPLOAD" { + handleUpload(c, m) + continue + } + if m.Type == "UPLOAD_BEGIN" { + handleUploadBegin(c, m) + continue + } + if m.Type == "UPLOAD_CHUNK" { + handleUploadChunk(c, m) + continue + } + if m.Type == "UPLOAD_END" { + handleUploadEnd(c, m) + continue + } + if m.Type == "DOWNLOAD_CHUNK" { + handleDownloadChunk(c, m) + continue + } + if m.Type == "DOWNLOAD" { + handleDownload(c, m) + continue + } + if m.Type != "SUBMIT" { + writeResult(c, farm.WorkerResult{Type: "FAILED", Error: "bad request"}) + continue + } + var j farm.RenderJob + if json.Unmarshal(m.Job, &j) != nil { + writeResult(c, farm.WorkerResult{Type: "FAILED", Error: "bad job"}) + continue + } + switch j.Engine { + case "null": + writeResult(c, farm.WorkerResult{ + Type: "COMPLETE", JobID: j.ID, Engine: "null", Output: j.Output, + }) + case "povray": + scene, release := leaseDownloadAsset(j.Scene) + if strings.HasPrefix(j.Scene, "asset-") && scene == "" { + writeResult(c, farm.WorkerResult{ + Type: "FAILED", JobID: j.ID, Engine: "povray", Error: "scene asset unavailable", + }) + continue + } + uploadSessions.Lock() + _, incomplete := uploadSessions.expected[filepath.Clean(scene)] + uploadSessions.Unlock() + if incomplete { + release() + writeResult(c, farm.WorkerResult{ + Type: "FAILED", JobID: j.ID, Engine: "povray", Error: "scene upload incomplete", + }) + continue + } + j.Scene = scene + func() { + defer release() + renderPOVRay(c, j) + }() + default: + writeResult(c, farm.WorkerResult{ + Type: "FAILED", JobID: j.ID, Engine: j.Engine, Error: "unsupported engine", + }) + } + } +} + +func validStagedAsset(asset string) (string, bool) { + if asset == "" || !filepath.IsAbs(asset) { + return "", false + } + clean := filepath.Clean(asset) + dir := filepath.Dir(clean) + if filepath.Dir(dir) != filepath.Clean(os.TempDir()) || !strings.HasPrefix(filepath.Base(dir), "amirender-asset-") { + return "", false + } + dirInfo, err := os.Lstat(dir) + if err != nil || !dirInfo.IsDir() || dirInfo.Mode()&os.ModeSymlink != 0 { + return "", false + } + fileInfo, err := os.Lstat(clean) + if err != nil || !fileInfo.Mode().IsRegular() { + return "", false + } + return clean, true +} + +// acquireAsset prevents TTL cleanup while a registered asset is in use. +func acquireAsset(id string) (string, func(), bool) { + assetIDs.Lock() + path, ok := assetIDs.paths[id] + if !ok { + assetIDs.Unlock() + return "", nil, false + } + assetIDs.active[id]++ + assetIDs.Unlock() + return path, func() { + assetIDs.Lock() + assetIDs.active[id]-- + if assetIDs.active[id] == 0 { + delete(assetIDs.active, id) + } + assetIDs.Unlock() + }, true +} + +// cleanupCompletedAssets revokes finished assets after their retention window. +func cleanupCompletedAssets(now time.Time) { + var expired []string + assetIDs.Lock() + for id, completedAt := range assetIDs.completed { + if now.Sub(completedAt) >= completedAssetTTL && assetIDs.active[id] == 0 { + expired = append(expired, assetIDs.paths[id]) + delete(assetIDs.paths, id) + delete(assetIDs.completed, id) + } + } + assetIDs.Unlock() + for _, path := range expired { + if _, ok := validStagedAsset(path); ok { + _ = os.RemoveAll(filepath.Dir(path)) + } + } +} + +func cleanupExpiredUploads(now time.Time) { + var expired []string + uploadSessions.Lock() + for path, session := range uploadSessions.expected { + if now.Sub(session.created) >= uploadSessionTTL { + delete(uploadSessions.expected, path) + expired = append(expired, session.dir) + } + } + uploadSessions.Unlock() + assetIDs.Lock() + for id, path := range assetIDs.paths { + for _, dir := range expired { + if filepath.Dir(path) == dir { + delete(assetIDs.paths, id) + break + } + } + } + assetIDs.Unlock() + for _, dir := range expired { + if dir == "" || !strings.HasPrefix(filepath.Base(dir), "amirender-asset-") { + continue + } + _ = os.RemoveAll(dir) + } +} + +func handleUploadBegin(c net.Conn, m message) { + cleanupExpiredUploads(time.Now()) + if filepath.Base(m.Name) != m.Name || m.Name == "." || m.Name == "" || m.Size <= 0 || m.Size > maxUploadBytes { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "invalid upload"}) + return + } + dir, err := os.MkdirTemp("", "amirender-asset-") + if err != nil { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "staging failed"}) + return + } + path := filepath.Join(dir, m.Name) + file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600) + if err != nil { + _ = os.RemoveAll(dir) + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "staging failed"}) + return + } + _ = file.Close() + id, err := newAssetID(path) + if err != nil { + _ = os.RemoveAll(dir) + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "asset registration failed"}) + return + } + uploadSessions.Lock() + uploadSessions.expected[path] = uploadSession{size: m.Size, created: time.Now(), dir: dir} + uploadSessions.Unlock() + _ = json.NewEncoder(c).Encode(stagedResult{Type: "STAGING", ID: id, Asset: path, Size: m.Size}) +} + +func handleUploadChunk(c net.Conn, m message) { + clean, ok := validStagedAsset(resolveUploadAsset(m.Asset)) + if !ok || m.Offset < 0 { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "unsafe asset reference"}) + return + } + data, err := base64.StdEncoding.DecodeString(m.Data) + if err != nil || len(data) == 0 || len(data) > 4096 || m.Offset+int64(len(data)) > maxUploadBytes { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "invalid chunk"}) + return + } + uploadSessions.Lock() + defer uploadSessions.Unlock() + session, exists := uploadSessions.expected[clean] + if !exists { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "upload session unavailable"}) + return + } + if m.Offset+int64(len(data)) > session.size { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "chunk exceeds declared upload size"}) + return + } + file, err := os.OpenFile(clean, os.O_WRONLY, 0600) + if err != nil { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "asset unavailable"}) + return + } + info, statErr := file.Stat() + if statErr != nil || info.Size() != m.Offset { + _ = file.Close() + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "unexpected chunk offset"}) + return + } + _, err = file.WriteAt(data, m.Offset) + _ = file.Close() + if err != nil { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "staging failed"}) + return + } + _ = json.NewEncoder(c).Encode(stagedResult{Type: "CHUNKED", Asset: clean, Offset: m.Offset + int64(len(data))}) +} + +func handleUploadEnd(c net.Conn, m message) { + clean, ok := validStagedAsset(resolveUploadAsset(m.Asset)) + if !ok || m.Size <= 0 || m.Size > maxUploadBytes { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "invalid upload completion"}) + return + } + uploadSessions.Lock() + defer uploadSessions.Unlock() + expected, exists := uploadSessions.expected[clean] + if !exists || expected.size != m.Size { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "upload size mismatch"}) + return + } + info, err := os.Stat(clean) + if err != nil { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "asset unavailable"}) + return + } + if info.Size() != m.Size { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "incomplete upload", Size: info.Size()}) + return + } + delete(uploadSessions.expected, clean) + markAssetCompleted(clean, time.Now()) + _ = json.NewEncoder(c).Encode(stagedResult{Type: "STAGED", Asset: clean, Size: info.Size()}) +} + +// leaseDownloadAsset protects ID-based downloads while they access the file. +// Legacy paths remain supported during the protocol migration. +func leaseDownloadAsset(asset string) (string, func()) { + if !strings.HasPrefix(asset, "asset-") { + return asset, func() {} + } + path, release, ok := acquireAsset(asset) + if !ok { + return "", func() {} + } + return path, release +} + +func handleDownloadChunk(c net.Conn, m message) { + path, release := leaseDownloadAsset(m.Asset) + defer release() + clean, ok := validStagedAsset(path) + if !ok || m.Offset < 0 || m.Size <= 0 || m.Size > 4096 { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "invalid chunk request"}) + return + } + file, err := os.Open(clean) + if err != nil { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "asset unavailable"}) + return + } + defer file.Close() + info, err := file.Stat() + if err != nil || m.Offset > info.Size() { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "invalid chunk offset"}) + return + } + buf := make([]byte, m.Size) + n, err := file.ReadAt(buf, m.Offset) + if err != nil && n == 0 { + if m.Offset == info.Size() { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "DATA", Asset: clean, Offset: m.Offset, Size: 0, EOF: true}) + return + } + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "asset unavailable"}) + return + } + _ = json.NewEncoder(c).Encode(stagedResult{Type: "DATA", Asset: clean, Offset: m.Offset, Size: int64(n), Data: base64.StdEncoding.EncodeToString(buf[:n]), EOF: m.Offset+int64(n) >= info.Size()}) +} + +func handleUpload(c net.Conn, m message) { + if filepath.Base(m.Name) != m.Name || m.Name == "." || m.Name == "" { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "unsafe asset name"}) + return + } + data, err := base64.StdEncoding.DecodeString(m.Data) + if err != nil || len(data) == 0 || len(data) > maxUploadBytes { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "invalid asset data"}) + return + } + dir, err := os.MkdirTemp("", "amirender-asset-") + if err != nil { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "staging failed"}) + return + } + path := filepath.Join(dir, m.Name) + if err := os.WriteFile(path, data, 0600); err != nil { + _ = os.RemoveAll(dir) + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "staging failed"}) + return + } + _ = json.NewEncoder(c).Encode(stagedResult{Type: "STAGED", Asset: path}) +} + +func handleDownload(c net.Conn, m message) { + path, release := leaseDownloadAsset(m.Asset) + defer release() + clean, ok := validStagedAsset(path) + if !ok { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "unsafe asset reference"}) + return + } + data, err := os.ReadFile(clean) + if err != nil || len(data) == 0 || len(data) > maxUploadBytes { + _ = json.NewEncoder(c).Encode(stagedResult{Type: "FAILED", Error: "asset unavailable"}) + return + } + _ = json.NewEncoder(c).Encode(stagedResult{ + Type: "DATA", Asset: clean, Data: base64.StdEncoding.EncodeToString(data), + }) +} + +func renderPOVRay(c net.Conn, j farm.RenderJob) { + dir, err := os.MkdirTemp("", "amirender-asset-") + if err != nil { + writeResult(c, farm.WorkerResult{ + Type: "FAILED", JobID: j.ID, Engine: "povray", Error: "output staging failed", + }) + return + } + output := filepath.Join(dir, "frame.png") + e := povray.New("") + result, err := e.Render(context.Background(), engine.Job{ + ID: j.ID, Scene: j.Scene, Output: output, Frame: j.Frame, Width: j.Width, Height: j.Height, + }) + if err != nil { + _ = os.RemoveAll(dir) + writeResult(c, farm.WorkerResult{ + Type: "FAILED", JobID: j.ID, Engine: "povray", Error: err.Error(), + }) + return + } + id, err := newAssetID(result.Output) + if err != nil { + _ = os.RemoveAll(dir) + writeResult(c, farm.WorkerResult{ + Type: "FAILED", JobID: j.ID, Engine: "povray", Error: "output registration failed", + }) + return + } + markAssetCompleted(result.Output, time.Now()) + writeResult(c, farm.WorkerResult{ + Type: "COMPLETE", JobID: j.ID, Engine: "povray", Output: result.Output, AssetID: id, + }) +} + +func writeResult(c net.Conn, result farm.WorkerResult) { + _ = json.NewEncoder(c).Encode(result) +} + +func Serve(ln net.Listener) error { + stop := make(chan struct{}) + defer close(stop) + go func() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + select { + case now := <-ticker.C: + cleanupExpiredUploads(now) + cleanupCompletedAssets(now) + case <-stop: + return + } + } + }() + for { + c, err := ln.Accept() + if err != nil { + return fmt.Errorf("accept: %w", err) + } + go Handle(c) + } +} diff --git a/internal/node/server_test.go b/internal/node/server_test.go new file mode 100644 index 0000000..3fb8c82 --- /dev/null +++ b/internal/node/server_test.go @@ -0,0 +1,1019 @@ +package node + +import ( + "bufio" + "context" + "encoding/base64" + "encoding/json" + "net" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/Ploos-AS/AmiRender/internal/farm" +) + +func testCoordinator(t *testing.T, engine string) (farm.Coordinator, func()) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + go func() { _ = Serve(ln) }() + + r := farm.NewRegistry() + if err := r.Register(farm.Node{ + ID: "node-01", Arch: "amd64", + Capabilities: []farm.Capability{{Engine: engine}}, + }); err != nil { + ln.Close() + t.Fatal(err) + } + return farm.Coordinator{ + Registry: r, + Endpoints: map[string]string{"node-01": ln.Addr().String()}, + Client: farm.TCPWorkerClient{}, + }, func() { _ = ln.Close() } +} + +func TestProductionHandlerNullRoundTrip(t *testing.T) { + c, closeServer := testCoordinator(t, "null") + defer closeServer() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + result, err := c.Render(ctx, farm.RenderJob{ + ID: "m1-e2e", Engine: "null", Output: "result.dat", + }) + if err != nil { + t.Fatal(err) + } + if result.Type != "COMPLETE" || result.JobID != "m1-e2e" || result.Output != "result.dat" { + t.Fatalf("unexpected result: %#v", result) + } +} + +func TestProductionHandlerPOVRayCreatesPNG(t *testing.T) { + if _, err := exec.LookPath("povray"); err != nil { + t.Skip("povray not installed") + } + + dir := t.TempDir() + scene := filepath.Join(dir, "scene.pov") + output := filepath.Join(dir, "frame.png") + source := `camera { location <0, 0, -3> look_at <0, 0, 0> } +light_source { <-2, 3, -4> color rgb <1, 1, 1> } +sphere { <0, 0, 0>, 1 pigment { color rgb <0.7, 0.7, 0.7> } } +` + if err := os.WriteFile(scene, []byte(source), 0600); err != nil { + t.Fatal(err) + } + + c, closeServer := testCoordinator(t, "povray") + defer closeServer() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + result, err := c.Render(ctx, farm.RenderJob{ + ID: "pov-e2e", Engine: "povray", Scene: scene, Output: output, Width: 160, Height: 120, + }) + if err != nil { + t.Fatal(err) + } + if result.Type != "COMPLETE" || result.Output == "" || result.Output == output { + t.Fatalf("unexpected result: %#v", result) + } + if !strings.HasPrefix(filepath.Base(filepath.Dir(result.Output)), "amirender-asset-") { + t.Fatalf("output is not worker-staged: %q", result.Output) + } + defer os.RemoveAll(filepath.Dir(result.Output)) + info, err := os.Stat(result.Output) + if err != nil { + t.Fatalf("render output missing: %v", err) + } + if info.Size() == 0 { + t.Fatal("render output is empty") + } +} + +func TestPOVRayUploadRenderDownloadRoundTrip(t *testing.T) { + if _, err := exec.LookPath("povray"); err != nil { + t.Skip("povray not installed") + } + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { _ = Serve(ln) }() + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + enc := json.NewEncoder(conn) + dec := json.NewDecoder(bufio.NewReader(conn)) + + source := `camera { location <0, 0, -3> look_at <0, 0, 0> } +light_source { <-2, 3, -4> color rgb <1, 1, 1> } +sphere { <0, 0, 0>, 1 pigment { color rgb <0.7, 0.7, 0.7> } } +` + if err := enc.Encode(map[string]string{ + "type": "UPLOAD", "name": "scene.pov", + "data": base64.StdEncoding.EncodeToString([]byte(source)), + }); err != nil { + t.Fatal(err) + } + var staged stagedResult + if err := dec.Decode(&staged); err != nil { + t.Fatal(err) + } + if staged.Type != "STAGED" || staged.Asset == "" { + t.Fatalf("unexpected staging result: %#v", staged) + } + defer os.RemoveAll(filepath.Dir(staged.Asset)) + + if err := enc.Encode(map[string]any{ + "type": "SUBMIT", + "job": map[string]any{ + "id": "pov-wire-e2e", "engine": "povray", "scene": staged.Asset, + "output": "RAM:frame.png", "width": 64, "height": 48, + }, + }); err != nil { + t.Fatal(err) + } + var rendered farm.WorkerResult + if err := dec.Decode(&rendered); err != nil { + t.Fatal(err) + } + if rendered.Type != "COMPLETE" || rendered.Output == "" || rendered.Output == "RAM:frame.png" { + t.Fatalf("unexpected render result: %#v", rendered) + } + defer os.RemoveAll(filepath.Dir(rendered.Output)) + + if err := enc.Encode(map[string]string{"type": "DOWNLOAD", "asset": rendered.Output}); err != nil { + t.Fatal(err) + } + var downloaded stagedResult + if err := dec.Decode(&downloaded); err != nil { + t.Fatal(err) + } + if downloaded.Type != "DATA" { + t.Fatalf("unexpected download result: %#v", downloaded) + } + png, err := base64.StdEncoding.DecodeString(downloaded.Data) + if err != nil { + t.Fatal(err) + } + signature := []byte{0x89, 'P', 'N', 'G', 0x0d, 0x0a, 0x1a, 0x0a} + if len(png) <= len(signature) || string(png[:len(signature)]) != string(signature) { + t.Fatalf("download is not a PNG: %x", png[:min(len(png), len(signature))]) + } +} + +func TestDownloadReturnsStagedAsset(t *testing.T) { + dir, err := os.MkdirTemp("", "amirender-asset-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + path := filepath.Join(dir, "frame.png") + want := []byte{0x89, 'P', 'N', 'G', 0x0d, 0x0a} + if err := os.WriteFile(path, want, 0600); err != nil { + t.Fatal(err) + } + + server, client := net.Pipe() + defer client.Close() + go Handle(server) + + enc := json.NewEncoder(client) + dec := json.NewDecoder(bufio.NewReader(client)) + if err := enc.Encode(map[string]string{"type": "DOWNLOAD", "asset": path}); err != nil { + t.Fatal(err) + } + var got stagedResult + if err := dec.Decode(&got); err != nil { + t.Fatal(err) + } + if got.Type != "DATA" || got.Asset != path { + t.Fatalf("unexpected download result: %#v", got) + } + decoded, err := base64.StdEncoding.DecodeString(got.Data) + if err != nil { + t.Fatal(err) + } + if string(decoded) != string(want) { + t.Fatalf("download mismatch: %v", decoded) + } + + if err := enc.Encode(map[string]string{"type": "DOWNLOAD", "asset": "/etc/passwd"}); err != nil { + t.Fatal(err) + } + var rejected stagedResult + if err := dec.Decode(&rejected); err != nil { + t.Fatal(err) + } + if rejected.Type != "FAILED" { + t.Fatalf("unsafe download was not rejected: %#v", rejected) + } +} + +func TestUploadStagesAssetAndRejectsTraversal(t *testing.T) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { _ = Serve(ln) }() + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + enc := json.NewEncoder(conn) + dec := json.NewDecoder(bufio.NewReader(conn)) + + data := base64.StdEncoding.EncodeToString([]byte("camera {}\\n")) + if err := enc.Encode(map[string]string{"type": "UPLOAD", "name": "scene.pov", "data": data}); err != nil { + t.Fatal(err) + } + var staged stagedResult + if err := dec.Decode(&staged); err != nil { + t.Fatal(err) + } + if staged.Type != "STAGED" || staged.Asset == "" { + t.Fatalf("unexpected staging result: %#v", staged) + } + defer os.RemoveAll(filepath.Dir(staged.Asset)) + got, err := os.ReadFile(staged.Asset) + if err != nil { + t.Fatal(err) + } + if string(got) != "camera {}\\n" { + t.Fatalf("unexpected staged data %q", got) + } + + if err := enc.Encode(map[string]string{"type": "UPLOAD", "name": "../escape.pov", "data": data}); err != nil { + t.Fatal(err) + } + var rejected stagedResult + if err := dec.Decode(&rejected); err != nil { + t.Fatal(err) + } + if rejected.Type != "FAILED" { + t.Fatalf("traversal upload was not rejected: %#v", rejected) + } +} + +func TestChunkedAssetRoundTripAndOffsetValidation(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + go Handle(server) + + enc := json.NewEncoder(client) + dec := json.NewDecoder(bufio.NewReader(client)) + want := make([]byte, 9000) + for i := range want { + want[i] = byte(i % 251) + } + + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_BEGIN", "name": "large.bin", "size": len(want), + }); err != nil { + t.Fatal(err) + } + var begun stagedResult + if err := dec.Decode(&begun); err != nil { + t.Fatal(err) + } + if begun.Type != "STAGING" || begun.Asset == "" || begun.Size != int64(len(want)) { + t.Fatalf("unexpected begin result: %#v", begun) + } + defer os.RemoveAll(filepath.Dir(begun.Asset)) + + offset := 0 + for offset < len(want) { + end := offset + 4096 + if end > len(want) { + end = len(want) + } + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": offset, + "data": base64.StdEncoding.EncodeToString(want[offset:end]), + }); err != nil { + t.Fatal(err) + } + var ack stagedResult + if err := dec.Decode(&ack); err != nil { + t.Fatal(err) + } + if ack.Type != "CHUNKED" || ack.Offset != int64(end) { + t.Fatalf("unexpected chunk acknowledgement: %#v", ack) + } + offset = end + } + + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": 1, + "data": base64.StdEncoding.EncodeToString([]byte("bad")), + }); err != nil { + t.Fatal(err) + } + var rejected stagedResult + if err := dec.Decode(&rejected); err != nil { + t.Fatal(err) + } + if rejected.Type != "FAILED" || rejected.Error != "unexpected chunk offset" { + t.Fatalf("bad offset was not rejected: %#v", rejected) + } + + var got []byte + for offset = 0; ; { + if err := enc.Encode(map[string]any{ + "type": "DOWNLOAD_CHUNK", "asset": begun.Asset, "offset": offset, "size": 4096, + }); err != nil { + t.Fatal(err) + } + var part stagedResult + if err := dec.Decode(&part); err != nil { + t.Fatal(err) + } + if part.Type != "DATA" || part.Offset != int64(offset) { + t.Fatalf("unexpected download chunk: %#v", part) + } + decoded, err := base64.StdEncoding.DecodeString(part.Data) + if err != nil { + t.Fatal(err) + } + if int64(len(decoded)) != part.Size { + t.Fatalf("chunk size mismatch: decoded=%d declared=%d", len(decoded), part.Size) + } + got = append(got, decoded...) + offset += len(decoded) + if part.EOF { + break + } + } + if string(got) != string(want) { + t.Fatalf("chunked round trip mismatch: got=%d want=%d", len(got), len(want)) + } + + if err := enc.Encode(map[string]any{ + "type": "DOWNLOAD_CHUNK", "asset": begun.Asset, "offset": 0, "size": 4097, + }); err != nil { + t.Fatal(err) + } + var oversized stagedResult + if err := dec.Decode(&oversized); err != nil { + t.Fatal(err) + } + if oversized.Type != "FAILED" { + t.Fatalf("oversized download chunk was not rejected: %#v", oversized) + } +} +func TestUploadEndRequiresDeclaredSizeAndSingleCompletion(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + go Handle(server) + enc := json.NewEncoder(client) + dec := json.NewDecoder(bufio.NewReader(client)) + send := func(m map[string]any) stagedResult { + t.Helper() + if err := enc.Encode(m); err != nil { + t.Fatal(err) + } + var result stagedResult + if err := dec.Decode(&result); err != nil { + t.Fatal(err) + } + return result + } + begun := send(map[string]any{"type": "UPLOAD_BEGIN", "name": "partial.bin", "size": 9000}) + if begun.Type != "STAGING" { + t.Fatalf("begin: %#v", begun) + } + defer os.RemoveAll(filepath.Dir(begun.Asset)) + payload := base64.StdEncoding.EncodeToString(make([]byte, 4096)) + for _, offset := range []int{0, 4096} { + ack := send(map[string]any{"type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": offset, "data": payload}) + if ack.Type != "CHUNKED" { + t.Fatalf("chunk: %#v", ack) + } + } + if result := send(map[string]any{"type": "UPLOAD_END", "asset": begun.Asset, "size": 8192}); result.Type != "FAILED" || result.Error != "upload size mismatch" { + t.Fatalf("changed declared size accepted: %#v", result) + } + if result := send(map[string]any{"type": "UPLOAD_END", "asset": begun.Asset, "size": 9000}); result.Type != "FAILED" || result.Error != "incomplete upload" { + t.Fatalf("partial upload accepted: %#v", result) + } + last := base64.StdEncoding.EncodeToString(make([]byte, 808)) + if result := send(map[string]any{"type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": 8192, "data": last}); result.Type != "CHUNKED" { + t.Fatalf("final chunk: %#v", result) + } + if result := send(map[string]any{"type": "UPLOAD_END", "asset": begun.Asset, "size": 9000}); result.Type != "STAGED" { + t.Fatalf("completion: %#v", result) + } + if result := send(map[string]any{"type": "UPLOAD_END", "asset": begun.Asset, "size": 9000}); result.Type != "FAILED" { + t.Fatalf("duplicate completion accepted: %#v", result) + } +} + +func TestPOVRayRejectsUnfinishedUpload(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + go Handle(server) + enc := json.NewEncoder(client) + dec := json.NewDecoder(bufio.NewReader(client)) + if err := enc.Encode(map[string]any{"type": "UPLOAD_BEGIN", "name": "pending.pov", "size": 11}); err != nil { + t.Fatal(err) + } + var begun stagedResult + if err := dec.Decode(&begun); err != nil { + t.Fatal(err) + } + if begun.Type != "STAGING" { + t.Fatalf("unexpected begin: %#v", begun) + } + defer os.RemoveAll(filepath.Dir(begun.Asset)) + job := farm.RenderJob{ID: "unfinished", Engine: "povray", Scene: begun.Asset, Output: "RAM:frame.png"} + if err := enc.Encode(map[string]any{"type": "SUBMIT", "job": job}); err != nil { + t.Fatal(err) + } + var rejected farm.WorkerResult + if err := dec.Decode(&rejected); err != nil { + t.Fatal(err) + } + if rejected.Type != "FAILED" || rejected.Error != "scene upload incomplete" { + t.Fatalf("unfinished upload was rendered: %#v", rejected) + } + payload := base64.StdEncoding.EncodeToString([]byte("camera {}")) + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": 0, "data": payload, + }); err != nil { + t.Fatal(err) + } + var chunk stagedResult + if err := dec.Decode(&chunk); err != nil { + t.Fatal(err) + } + if chunk.Type != "CHUNKED" || chunk.Offset != 9 { + t.Fatalf("unexpected chunk: %#v", chunk) + } + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": 9, + "data": base64.StdEncoding.EncodeToString([]byte("\\n")), + }); err != nil { + t.Fatal(err) + } + if err := dec.Decode(&chunk); err != nil { + t.Fatal(err) + } + if chunk.Type != "CHUNKED" || chunk.Offset != 11 { + t.Fatalf("unexpected final chunk: %#v", chunk) + } + if err := enc.Encode(map[string]any{"type": "UPLOAD_END", "asset": begun.Asset, "size": 11}); err != nil { + t.Fatal(err) + } + var staged stagedResult + if err := dec.Decode(&staged); err != nil { + t.Fatal(err) + } + if staged.Type != "STAGED" { + t.Fatalf("completed upload rejected: %#v", staged) + } + uploadSessions.Lock() + _, pending := uploadSessions.expected[begun.Asset] + uploadSessions.Unlock() + if pending { + t.Fatal("completed upload remains blocked") + } +} + +func TestCleanupExpiredUploadsPreservesActiveSessions(t *testing.T) { + now := time.Now() + root := t.TempDir() + expiredDir, err := os.MkdirTemp(root, "amirender-asset-") + if err != nil { + t.Fatal(err) + } + activeDir, err := os.MkdirTemp(root, "amirender-asset-") + if err != nil { + t.Fatal(err) + } + expiredPath := filepath.Join(expiredDir, "old.pov") + activePath := filepath.Join(activeDir, "new.pov") + for _, path := range []string{expiredPath, activePath} { + if err := os.WriteFile(path, []byte("scene"), 0600); err != nil { + t.Fatal(err) + } + } + uploadSessions.Lock() + uploadSessions.expected[expiredPath] = uploadSession{size: 5, created: now.Add(-uploadSessionTTL - time.Second), dir: expiredDir} + uploadSessions.expected[activePath] = uploadSession{size: 5, created: now.Add(-uploadSessionTTL + time.Second), dir: activeDir} + uploadSessions.Unlock() + defer func() { + uploadSessions.Lock() + delete(uploadSessions.expected, expiredPath) + delete(uploadSessions.expected, activePath) + uploadSessions.Unlock() + }() + cleanupExpiredUploads(now) + uploadSessions.Lock() + _, expiredExists := uploadSessions.expected[expiredPath] + _, activeExists := uploadSessions.expected[activePath] + uploadSessions.Unlock() + if expiredExists || !activeExists { + t.Fatalf("unexpected sessions: expired=%t active=%t", expiredExists, activeExists) + } + if _, err := os.Stat(expiredPath); !os.IsNotExist(err) { + t.Fatalf("expired file still exists: %v", err) + } + if _, err := os.Stat(activePath); err != nil { + t.Fatalf("active file removed: %v", err) + } +} + +func TestCleanupExpiredUploadsDoesNotDeleteUnregisteredDirectory(t *testing.T) { + now := time.Now() + root := t.TempDir() + unregistered := filepath.Join(root, "other-project") + if err := os.Mkdir(unregistered, 0700); err != nil { + t.Fatal(err) + } + path := filepath.Join(unregistered, "keep.txt") + if err := os.WriteFile(path, []byte("keep"), 0600); err != nil { + t.Fatal(err) + } + uploadSessions.Lock() + uploadSessions.expected[path] = uploadSession{ + size: 4, created: now.Add(-uploadSessionTTL - time.Second), dir: unregistered, + } + uploadSessions.Unlock() + defer func() { + uploadSessions.Lock() + delete(uploadSessions.expected, path) + uploadSessions.Unlock() + }() + cleanupExpiredUploads(now) + if _, err := os.Stat(path); err != nil { + t.Fatalf("unregistered directory was removed: %v", err) + } +} + +func TestUploadChunkEnforcesDeclaredSizeAndFinalization(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + go Handle(server) + enc := json.NewEncoder(client) + dec := json.NewDecoder(bufio.NewReader(client)) + send := func(m map[string]any) stagedResult { + t.Helper() + if err := enc.Encode(m); err != nil { + t.Fatal(err) + } + var result stagedResult + if err := dec.Decode(&result); err != nil { + t.Fatal(err) + } + return result + } + begun := send(map[string]any{"type": "UPLOAD_BEGIN", "name": "size.bin", "size": 3}) + if begun.Type != "STAGING" { + t.Fatalf("begin: %#v", begun) + } + defer os.RemoveAll(filepath.Dir(begun.Asset)) + oversized := send(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": 0, + "data": base64.StdEncoding.EncodeToString([]byte("four")), + }) + if oversized.Type != "FAILED" || oversized.Error != "chunk exceeds declared upload size" { + t.Fatalf("oversized chunk accepted: %#v", oversized) + } + if result := send(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": 0, + "data": base64.StdEncoding.EncodeToString([]byte("abc")), + }); result.Type != "CHUNKED" || result.Offset != 3 { + t.Fatalf("valid chunk rejected: %#v", result) + } + if result := send(map[string]any{"type": "UPLOAD_END", "asset": begun.Asset, "size": 3}); result.Type != "STAGED" { + t.Fatalf("completion rejected: %#v", result) + } + if result := send(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": begun.Asset, "offset": 3, + "data": base64.StdEncoding.EncodeToString([]byte("x")), + }); result.Type != "FAILED" || result.Error != "upload session unavailable" { + t.Fatalf("chunk after completion accepted: %#v", result) + } +} + +func TestUploadEndAndCleanupAreSerialized(t *testing.T) { + now := time.Now() + dir, err := os.MkdirTemp("", "amirender-asset-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + path := filepath.Join(dir, "race.pov") + if err := os.WriteFile(path, []byte("abc"), 0600); err != nil { + t.Fatal(err) + } + uploadSessions.Lock() + uploadSessions.expected[path] = uploadSession{ + size: 3, created: now.Add(-uploadSessionTTL - time.Second), dir: dir, + } + uploadSessions.Unlock() + defer func() { + uploadSessions.Lock() + delete(uploadSessions.expected, path) + uploadSessions.Unlock() + }() + server, client := net.Pipe() + defer client.Close() + go Handle(server) + enc := json.NewEncoder(client) + dec := json.NewDecoder(bufio.NewReader(client)) + uploadSessions.Lock() + done := make(chan struct{}) + go func() { + defer close(done) + cleanupExpiredUploads(now) + }() + if err := enc.Encode(map[string]any{"type": "UPLOAD_END", "asset": path, "size": 3}); err != nil { + uploadSessions.Unlock() + t.Fatal(err) + } + uploadSessions.Unlock() + var result stagedResult + if err := dec.Decode(&result); err != nil { + t.Fatal(err) + } + <-done + if result.Type != "STAGED" && result.Type != "FAILED" { + t.Fatalf("unexpected concurrent completion: %#v", result) + } + if result.Type == "STAGED" { + if _, err := os.Stat(path); err != nil { + t.Fatalf("completed upload deleted by cleanup: %v", err) + } + } +} + +func TestValidStagedAssetRejectsSymlinksAndOutsideTemp(t *testing.T) { + root := t.TempDir() + dir := filepath.Join(root, "amirender-asset-test") + if err := os.Mkdir(dir, 0700); err != nil { + t.Fatal(err) + } + file := filepath.Join(dir, "scene.pov") + if err := os.WriteFile(file, []byte("scene"), 0600); err != nil { + t.Fatal(err) + } + if _, ok := validStagedAsset(file); ok { + t.Fatal("asset outside os.TempDir accepted") + } + + tempDir, err := os.MkdirTemp("", "amirender-asset-test-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(tempDir) + target := filepath.Join(tempDir, "real.pov") + if err := os.WriteFile(target, []byte("scene"), 0600); err != nil { + t.Fatal(err) + } + link := filepath.Join(tempDir, "link.pov") + if err := os.Symlink(target, link); err != nil { + t.Skipf("symlinks unavailable: %v", err) + } + if _, ok := validStagedAsset(link); ok { + t.Fatal("symlink asset accepted") + } + if _, ok := validStagedAsset(target); !ok { + t.Fatal("valid staging asset rejected") + } +} + +func TestOpaqueAssetIDChunkedUpload(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + go Handle(server) + enc := json.NewEncoder(client) + dec := json.NewDecoder(client) + payload := []byte("camera {}\n") + if err := enc.Encode(map[string]any{"type": "UPLOAD_BEGIN", "name": "opaque.pov", "size": len(payload)}); err != nil { + t.Fatal(err) + } + var begun stagedResult + if err := dec.Decode(&begun); err != nil { + t.Fatal(err) + } + if begun.Type != "STAGING" || !strings.HasPrefix(begun.ID, "asset-") { + t.Fatalf("missing opaque asset ID: %#v", begun) + } + defer os.RemoveAll(filepath.Dir(begun.Asset)) + if err := enc.Encode(map[string]any{"type": "UPLOAD_CHUNK", "asset": begun.ID, "offset": 0, "data": base64.StdEncoding.EncodeToString(payload)}); err != nil { + t.Fatal(err) + } + var chunk stagedResult + if err := dec.Decode(&chunk); err != nil { + t.Fatal(err) + } + if chunk.Type != "CHUNKED" || chunk.Offset != int64(len(payload)) { + t.Fatalf("opaque chunk failed: %#v", chunk) + } + if err := enc.Encode(map[string]any{"type": "UPLOAD_END", "asset": begun.ID, "size": len(payload)}); err != nil { + t.Fatal(err) + } + var end stagedResult + if err := dec.Decode(&end); err != nil { + t.Fatal(err) + } + if end.Type != "STAGED" { + t.Fatalf("opaque finalization failed: %#v", end) + } + data, err := os.ReadFile(begun.Asset) + if err != nil || string(data) != string(payload) { + t.Fatalf("uploaded asset mismatch: %q, %v", data, err) + } + if err := enc.Encode(map[string]any{"type": "UPLOAD_CHUNK", "asset": "asset-unknown", "offset": 0, "data": base64.StdEncoding.EncodeToString(payload)}); err != nil { + t.Fatal(err) + } + var rejected stagedResult + if err := dec.Decode(&rejected); err != nil { + t.Fatal(err) + } + if rejected.Type != "FAILED" { + t.Fatalf("unknown opaque asset ID accepted: %#v", rejected) + } +} + +func TestOpaqueAssetIDDownload(t *testing.T) { + server, client := net.Pipe() + defer client.Close() + go Handle(server) + enc := json.NewEncoder(client) + dec := json.NewDecoder(client) + payload := []byte("opaque download") + if err := enc.Encode(map[string]any{"type": "UPLOAD_BEGIN", "name": "download.bin", "size": len(payload)}); err != nil { + t.Fatal(err) + } + var begun stagedResult + if err := dec.Decode(&begun); err != nil { + t.Fatal(err) + } + if begun.Type != "STAGING" || begun.ID == "" { + t.Fatalf("missing asset ID: %#v", begun) + } + defer os.RemoveAll(filepath.Dir(begun.Asset)) + if err := enc.Encode(map[string]any{"type": "UPLOAD_CHUNK", "asset": begun.ID, "offset": 0, "data": base64.StdEncoding.EncodeToString(payload)}); err != nil { + t.Fatal(err) + } + var chunk stagedResult + if err := dec.Decode(&chunk); err != nil { + t.Fatal(err) + } + if chunk.Type != "CHUNKED" { + t.Fatalf("chunk upload failed: %#v", chunk) + } + if err := enc.Encode(map[string]any{"type": "UPLOAD_END", "asset": begun.ID, "size": len(payload)}); err != nil { + t.Fatal(err) + } + var completed stagedResult + if err := dec.Decode(&completed); err != nil { + t.Fatal(err) + } + if completed.Type != "STAGED" { + t.Fatalf("finalization failed: %#v", completed) + } + for _, typ := range []string{"DOWNLOAD", "DOWNLOAD_CHUNK"} { + req := map[string]any{"type": typ, "asset": begun.ID} + if typ == "DOWNLOAD_CHUNK" { + req["offset"] = 0 + req["size"] = 4096 + } + if err := enc.Encode(req); err != nil { + t.Fatal(err) + } + var result stagedResult + if err := dec.Decode(&result); err != nil { + t.Fatal(err) + } + data, err := base64.StdEncoding.DecodeString(result.Data) + if err != nil || result.Type != "DATA" || string(data) != string(payload) { + t.Fatalf("%s failed: %#v, %v", typ, result, err) + } + } + if err := enc.Encode(map[string]any{"type": "DOWNLOAD", "asset": "asset-unknown"}); err != nil { + t.Fatal(err) + } + var rejected stagedResult + if err := dec.Decode(&rejected); err != nil { + t.Fatal(err) + } + if rejected.Type != "FAILED" { + t.Fatalf("unknown asset ID accepted: %#v", rejected) + } +} + +func TestCleanupExpiredUploadsRevokesAssetIDs(t *testing.T) { + dir, err := os.MkdirTemp("", "amirender-asset-expired-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + path := filepath.Join(dir, "expired.pov") + if err := os.WriteFile(path, []byte("scene"), 0600); err != nil { + t.Fatal(err) + } + id, err := newAssetID(path) + if err != nil { + t.Fatal(err) + } + now := time.Now() + uploadSessions.Lock() + uploadSessions.expected[path] = uploadSession{size: 5, created: now.Add(-uploadSessionTTL - time.Minute), dir: dir} + uploadSessions.Unlock() + cleanupExpiredUploads(now) + if resolved := resolveUploadAsset(id); resolved != "" { + t.Fatalf("expired ID still resolves: %q", resolved) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("expired staging directory still exists: %v", err) + } +} + +func TestCompletedAssetRetentionTTL(t *testing.T) { + dir, err := os.MkdirTemp("", "amirender-asset-retention-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + path := filepath.Join(dir, "scene.pov") + if err := os.WriteFile(path, []byte("scene"), 0600); err != nil { + t.Fatal(err) + } + id, err := newAssetID(path) + if err != nil { + t.Fatal(err) + } + now := time.Now() + markAssetCompleted(path, now) + cleanupCompletedAssets(now.Add(completedAssetTTL - time.Second)) + if got := resolveUploadAsset(id); got != path { + t.Fatalf("completed asset expired too early: %q", got) + } + if _, err := os.Stat(path); err != nil { + t.Fatalf("retained asset missing: %v", err) + } + cleanupCompletedAssets(now.Add(completedAssetTTL)) + if got := resolveUploadAsset(id); got != "" { + t.Fatalf("expired completed asset still resolves: %q", got) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("expired completed asset directory still exists: %v", err) + } +} + +func TestCompletedAssetLeaseDefersCleanup(t *testing.T) { + dir, err := os.MkdirTemp("", "amirender-asset-leased-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + path := filepath.Join(dir, "scene.pov") + if err := os.WriteFile(path, []byte("scene"), 0600); err != nil { + t.Fatal(err) + } + id, err := newAssetID(path) + if err != nil { + t.Fatal(err) + } + now := time.Now() + markAssetCompleted(path, now) + leasedPath, release, ok := acquireAsset(id) + if !ok || leasedPath != path { + t.Fatalf("failed to acquire asset lease: %q", leasedPath) + } + cleanupCompletedAssets(now.Add(completedAssetTTL + time.Second)) + if got := resolveUploadAsset(id); got != path { + t.Fatalf("active asset revoked: %q", got) + } + if _, err := os.Stat(path); err != nil { + t.Fatalf("leased asset deleted: %v", err) + } + release() + cleanupCompletedAssets(now.Add(completedAssetTTL + time.Second)) + if got := resolveUploadAsset(id); got != "" { + t.Fatalf("released expired asset still resolves: %q", got) + } + if _, err := os.Stat(dir); !os.IsNotExist(err) { + t.Fatalf("released expired asset directory still exists: %v", err) + } +} + +func TestPOVRayOpaqueIDRenderRoundTrip(t *testing.T) { + if _, err := exec.LookPath("povray"); err != nil { + t.Skip("povray not installed") + } + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go func() { _ = Serve(ln) }() + + conn, err := net.Dial("tcp", ln.Addr().String()) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + enc := json.NewEncoder(conn) + dec := json.NewDecoder(bufio.NewReader(conn)) + + source := `camera { location <0, 0, -3> look_at <0, 0, 0> } +light_source { <-2, 3, -4> color rgb <1, 1, 1> } +sphere { <0, 0, 0>, 1 pigment { color rgb <0.7, 0.7, 0.7> } } +` + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_BEGIN", "name": "scene.pov", "size": len(source), + }); err != nil { + t.Fatal(err) + } + var staged stagedResult + if err := dec.Decode(&staged); err != nil { + t.Fatal(err) + } + if staged.Type != "STAGING" || staged.Asset == "" || staged.ID == "" { + t.Fatalf("unexpected staging result: %#v", staged) + } + defer os.RemoveAll(filepath.Dir(staged.Asset)) + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_CHUNK", "asset": staged.ID, "offset": 0, + "data": base64.StdEncoding.EncodeToString([]byte(source)), + }); err != nil { + t.Fatal(err) + } + var chunk stagedResult + if err := dec.Decode(&chunk); err != nil { + t.Fatal(err) + } + if chunk.Type != "CHUNKED" { + t.Fatalf("unexpected chunk result: %#v", chunk) + } + if err := enc.Encode(map[string]any{ + "type": "UPLOAD_END", "asset": staged.ID, "size": len(source), + }); err != nil { + t.Fatal(err) + } + var completed stagedResult + if err := dec.Decode(&completed); err != nil { + t.Fatal(err) + } + if completed.Type != "STAGED" { + t.Fatalf("unexpected completion result: %#v", completed) + } + + if err := enc.Encode(map[string]any{ + "type": "SUBMIT", + "job": map[string]any{ + "id": "pov-wire-e2e", "engine": "povray", "scene": staged.ID, + "output": "RAM:frame.png", "width": 64, "height": 48, + }, + }); err != nil { + t.Fatal(err) + } + var rendered farm.WorkerResult + if err := dec.Decode(&rendered); err != nil { + t.Fatal(err) + } + if rendered.Type != "COMPLETE" || rendered.Output == "" || rendered.Output == "RAM:frame.png" || !strings.HasPrefix(rendered.AssetID, "asset-") { + t.Fatalf("unexpected render result: %#v", rendered) + } + defer os.RemoveAll(filepath.Dir(rendered.Output)) + + if err := enc.Encode(map[string]string{"type": "DOWNLOAD", "asset": rendered.AssetID}); err != nil { + t.Fatal(err) + } + var downloaded stagedResult + if err := dec.Decode(&downloaded); err != nil { + t.Fatal(err) + } + if downloaded.Type != "DATA" { + t.Fatalf("unexpected download result: %#v", downloaded) + } + png, err := base64.StdEncoding.DecodeString(downloaded.Data) + if err != nil { + t.Fatal(err) + } + signature := []byte{0x89, 'P', 'N', 'G', 0x0d, 0x0a, 0x1a, 0x0a} + if len(png) <= len(signature) || string(png[:len(signature)]) != string(signature) { + t.Fatalf("download is not a PNG: %x", png[:min(len(png), len(signature))]) + } +} diff --git a/protocol/messages.md b/protocol/messages.md new file mode 100644 index 0000000..a8a1058 --- /dev/null +++ b/protocol/messages.md @@ -0,0 +1,129 @@ +# Control messages + +AmiRender control messages are newline-delimited JSON. + +## REGISTER + +A worker registers itself with the controller and advertises render capabilities. + +Example: + +```json +{ + "type": "REGISTER", + "node": { + "version": 1, + "id": "rock5-01", + "arch": "arm64", + "os": "linux", + "memory_mb": 16384, + "capabilities": [ + {"engine": "povray"} + ] + } +} +``` + +Re-registering the same node ID replaces its previous capability record. This allows a worker to update capabilities after configuration or runtime changes. + +## REGISTERED + +The controller acknowledges a valid registration: + +```json +{"type":"REGISTERED","node_id":"rock5-01"} +``` + +Registration is the M1 discovery baseline. Network service discovery such as mDNS may locate controllers/workers later, but capability truth comes from the registration protocol. + + +## UPLOAD + +A client may stage one small render asset on a worker before submitting a job. +M1 uses base64 inside the existing newline-delimited JSON framing. + +```json +{"type":"UPLOAD","name":"scene.pov","data":"Y2FtZXJhIHt9Cg=="} +``` + +M1 constraints: + +- decoded payload size is limited to 1 MiB; +- `name` is a basename only; directory traversal and client-selected worker paths are rejected; +- the worker creates an isolated staging directory and chooses the filesystem path; +- staged files are private to the worker and are not a durable asset store; +- M1 initially targets a self-contained single-file POV-Ray scene. Includes, textures and multi-file bundles require a later asset-manifest protocol. + +## STAGED + +A successful upload returns the worker-selected asset reference: + +```json +{"type":"STAGED","asset":"/worker-selected/staging/path/scene.pov"} +``` + +The path is an opaque M1 asset reference from the client's perspective. Clients must not construct, modify, or predict it. A following `SUBMIT` may use this returned reference as its scene value. + +Invalid names, invalid base64, empty data, oversized data, or staging failures return `FAILED`. + + +## SUBMIT and remote output + +A remote render job may use a staged asset reference as its scene. The client's requested output path is a client-side destination hint; it is not a filesystem path that a Linux worker may write directly. + +For engines that produce files remotely, the worker creates its own staging directory and returns the staged result in `COMPLETE.output`. + +Example: + +```json +{"type":"SUBMIT","job":{"id":"frame-1","engine":"povray","scene":"","output":"RAM:frame.png","width":320,"height":256}} +``` + +```json +{"type":"COMPLETE","job_id":"frame-1","engine":"povray","output":""} +``` + +Clients must treat `COMPLETE.output` as an opaque worker asset reference in remote mode. In particular, a worker must not interpret an Amiga path such as `RAM:frame.png` or `DH0:...` as a local Linux path. + +## DOWNLOAD + +A client retrieves a staged result by sending the asset reference returned by the worker: + +```json +{"type":"DOWNLOAD","asset":""} +``` + +M1 uses the same bounded base64 transfer model as `UPLOAD`. The requested asset must be a worker-controlled staged asset; arbitrary filesystem paths are rejected. + +## DATA + +A successful download returns: + +```json +{"type":"DATA","asset":"","data":""} +``` + +The client decodes `data` and writes it to its own requested local output path. This keeps Amiga filesystem semantics on the Amiga side and Linux filesystem semantics on the worker side. + +## M1 end-to-end contract + +The qualified single-file POV-Ray flow is: + +```text +local scene + -> UPLOAD + <- STAGED + -> SUBMIT + <- COMPLETE + -> DOWNLOAD + <- DATA + -> local output +``` + +CI exercises this sequence over one TCP connection with a real POV-Ray render and verifies that the downloaded result has a valid PNG signature. + +## Known M1 transfer limitations + +The current base64-in-JSON transfer is deliberately simple and bounded. It is suitable for qualification and small assets, not production-size scenes or render outputs. + +A later protocol revision should add chunked upload/download, opaque asset IDs rather than filesystem paths, explicit asset sizes/hashes, multi-file manifests, and staged-asset lifetime/cleanup semantics. diff --git a/protocol/node.schema.json b/protocol/node.schema.json index adf2a65..0d685ac 100644 --- a/protocol/node.schema.json +++ b/protocol/node.schema.json @@ -2,17 +2,28 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "title": "AmiRender Node", "type": "object", - "required": ["version", "id", "arch", "engines"], + "required": ["version", "id", "arch", "capabilities"], "properties": { "version": { "type": "integer", "const": 1 }, "id": { "type": "string", "minLength": 1 }, "arch": { "type": "string", "enum": ["m68k", "arm64", "amd64"] }, "os": { "type": "string" }, "memory_mb": { "type": "integer", "minimum": 1 }, - "engines": { + "capabilities": { "type": "array", - "items": { "type": "string", "minLength": 1 }, - "uniqueItems": true + "items": { + "type": "object", + "required": ["engine"], + "properties": { + "engine": { "type": "string", "minLength": 1 }, + "modes": { + "type": "array", + "items": { "type": "string", "minLength": 1 }, + "uniqueItems": true + } + }, + "additionalProperties": false + } } }, "additionalProperties": false diff --git a/tests/scenes/m1-smoke.pov b/tests/scenes/m1-smoke.pov new file mode 100644 index 0000000..b1b2e08 --- /dev/null +++ b/tests/scenes/m1-smoke.pov @@ -0,0 +1,4 @@ +// AmiRender M1 POV-Ray smoke scene +camera { location <0, 0, -3> look_at <0, 0, 0> } +light_source { <-2, 3, -4> color rgb <1, 1, 1> } +sphere { <0, 0, 0>, 1 pigment { color rgb <0.7, 0.7, 0.7> } }