feat(enterprise): add scheduler reload and trigger commands

README.md

@@ -135,12 +135,16 @@ arcli compaction status
 arcli retention list
 arcli backup create --wait
 arcli logs --level warn --since 6h
+arcli scheduler cq reload
+arcli scheduler retention trigger
 ```
 
 `backup restore` is destructive and leaves metadata restoration to Arc's defaults: enabled on standalone nodes when present in the backup, skipped on cluster nodes. Use `--with-metadata` to request it explicitly (clusters refuse it), `--data-only` to explicitly skip metadata and config, or `--with-config` to also request config restoration. Stop writers and compaction before restoring. When the server reports staged metadata or a required restart, wait for the restore to complete before restarting.
 
 The command tree also covers API tokens, continuous queries, schedulers, predicate deletes, backups, cluster membership, and compaction. Run `arcli --help` or `arcli <command> --help` for the complete command reference.
 
+`scheduler cq reload` refreshes all active continuous queries in the scheduler; use `cq execute` to run one query immediately. `scheduler retention trigger` runs every active retention policy and may permanently delete old data, so it prompts for confirmation; pass `--yes` for scripted use. Use `retention execute <id|name>` to run one selected policy. Scheduler errors preserve Arc's status and message: HTTP 403 for missing entitlement, 503 for a licensed but unavailable scheduler, and 409 for retention role restrictions or an already-running operation.
+
 `license status` reads public `/health` without sending a token or checking authentication. It works on standalone and clustered deployments and supports table and JSON output. It reports the server's tier, status, source, and optional expiry, days remaining, and site-license state. Expiry is normalized to UTC RFC3339 with fractional seconds when provided.
 
 Active, expired, invalid, and configured-but-unlicensed states retain the server's status. A missing or null license block is normal for OSS and is distinct from `unlicensed`: JSON returns `{"configured": false, "license": null}`. Present licenses use `{"configured": true, "license": {...}}`; optional fields are omitted when absent or null, while explicit zero days and `false` site-license values are retained. Unknown server status values are preserved. Successfully reading any of these states exits 0; configuration, request, decoding, and output failures exit nonzero. Installing licenses or configuring license keys/files belongs to the Arc server.

docs/releases/unreleased.md

@@ -1,5 +1,7 @@
 ### Added
 
+- `arcli scheduler cq reload` refreshes all active continuous queries, and `arcli scheduler retention trigger` runs all active retention policies with a confirmation prompt. Both preserve Arc's error status and message, including HTTP 403 for missing scheduler entitlement, 503 for a licensed but unavailable scheduler, and 409 for retention role restrictions or an already-running operation. Closes [arcli#33](https://github.com/Basekick-Labs/arcli/issues/33).
+
 - `arcli license status` inspects the public `/health` license block on standalone and clustered Arc deployments, with table and JSON output. Reports tier, server status, source, and optional expiry, days remaining, and site-license state. Expiry uses UTC RFC3339 with fractional seconds when provided.
 - Absent/null license blocks succeed as normal OSS behavior and produce `{"configured":false,"license":null}` in JSON. Present licenses produce `{"configured":true,"license":{...}}`, preserving active, expired, invalid, unlicensed, and unknown server statuses. Optional absent/null fields are omitted; explicit zero days and false site-license values are retained.
 - All successfully read license states exit 0. Configuration, request, decoding, and output failures exit nonzero. No token or database header is sent to `/health`, and no authentication verification is performed. Normal connection selection, timeout, TLS verification, and identity opt-out settings apply.

internal/client/scheduler.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
 )
 
 // SchedulerBlock is one of the two entries of GET /api/v1/schedulers/.
@@ -61,3 +63,48 @@ func (c *Client) SchedulerStatus(ctx context.Context) (*SchedulersStatus, error)
 	out.Raw = body
 	return &out, nil
 }
+
+// SchedulerActionResult is the response from a scheduler administration
+// endpoint. JobCount is present only for CQ reloads.
+type SchedulerActionResult struct {
+	Message  string          `json:"message"`
+	JobCount *int            `json:"job_count,omitempty"`
+	Raw      json.RawMessage `json:"-"`
+}
+
+// ReloadCQScheduler reloads every active continuous query into the scheduler.
+func (c *Client) ReloadCQScheduler(ctx context.Context) (*SchedulerActionResult, error) {
+	return c.schedulerAction(ctx, "/api/v1/schedulers/cq/reload", "reload CQ scheduler")
+}
+
+// TriggerRetentionScheduler runs every active retention policy immediately.
+func (c *Client) TriggerRetentionScheduler(ctx context.Context) (*SchedulerActionResult, error) {
+	return c.schedulerAction(ctx, "/api/v1/schedulers/retention/trigger", "trigger retention scheduler")
+}
+
+func (c *Client) schedulerAction(ctx context.Context, path, action string) (*SchedulerActionResult, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.cfg.Endpoint+path, nil)
+	if err != nil {
+		return nil, fmt.Errorf("build request: %w", err)
+	}
+	req.Header.Set("Accept", "application/json")
+	c.setCrossDBHeaders(req)
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("%s: %w", action, err)
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+	if err != nil {
+		return nil, fmt.Errorf("read %s response: %w", action, err)
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return nil, decodeWriteError(resp.StatusCode, body)
+	}
+	var out SchedulerActionResult
+	if err := json.Unmarshal(body, &out); err != nil {
+		return nil, fmt.Errorf("decode %s response: %w", action, err)
+	}
+	out.Raw = body
+	return &out, nil
+}

internal/client/scheduler_test.go

@@ -0,0 +1,75 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"strings"
+	"testing"
+)
+
+func TestSchedulerActions_Success(t *testing.T) {
+	cli, _ := newAuthTestClient(t, func(w http.ResponseWriter, r *http.Request) {
+		assertAuthHeaders(t, r)
+		if r.Method != http.MethodPost || r.ContentLength != 0 {
+			t.Errorf("request method=%s body length=%d", r.Method, r.ContentLength)
+		}
+		switch r.URL.Path {
+		case "/api/v1/schedulers/cq/reload":
+			_, _ = w.Write([]byte(`{"message":"CQ scheduler reloaded successfully","job_count":3,"extra":"kept"}`))
+		case "/api/v1/schedulers/retention/trigger":
+			_, _ = w.Write([]byte(`{"message":"Retention triggered successfully","extra":"kept"}`))
+		default:
+			t.Errorf("unexpected path %s", r.URL.Path)
+		}
+	})
+	ctx := context.Background()
+	reload, err := cli.ReloadCQScheduler(ctx)
+	if err != nil || reload.Message != "CQ scheduler reloaded successfully" || reload.JobCount == nil || *reload.JobCount != 3 || !strings.Contains(string(reload.Raw), `"extra":"kept"`) {
+		t.Errorf("reload=%+v err=%v", reload, err)
+	}
+	trigger, err := cli.TriggerRetentionScheduler(ctx)
+	if err != nil || trigger.Message != "Retention triggered successfully" || trigger.JobCount != nil || !strings.Contains(string(trigger.Raw), `"extra":"kept"`) {
+		t.Errorf("trigger=%+v err=%v", trigger, err)
+	}
+}
+
+func TestSchedulerActions_PreserveDistinctArcErrors(t *testing.T) {
+	actions := []struct {
+		name string
+		call func(*Client) error
+	}{
+		{"CQ reload", func(c *Client) error { _, err := c.ReloadCQScheduler(context.Background()); return err }},
+		{"retention trigger", func(c *Client) error { _, err := c.TriggerRetentionScheduler(context.Background()); return err }},
+	}
+	cases := []struct {
+		name    string
+		status  int
+		message string
+	}{
+		{"admin permission", http.StatusForbidden, "Permission denied: admin permission required"},
+		{"missing entitlement", http.StatusForbidden, "Enterprise license required for scheduler controls"},
+		{"licensed scheduler unavailable", http.StatusServiceUnavailable, "Scheduler is not running"},
+		{"retention role restriction", http.StatusConflict, "Retention trigger failed: retention: node role is not primary writer"},
+		{"already running", http.StatusConflict, "Scheduler operation is already running"},
+	}
+	for _, action := range actions {
+		for _, tc := range cases {
+			t.Run(action.name+"/"+tc.name, func(t *testing.T) {
+				cli, _ := newAuthTestClient(t, func(w http.ResponseWriter, r *http.Request) {
+					w.WriteHeader(tc.status)
+					_ = json.NewEncoder(w).Encode(map[string]string{"error": tc.message})
+				})
+				err := action.call(cli)
+				var he *HTTPError
+				if !errors.As(err, &he) {
+					t.Fatalf("error type = %T, want *HTTPError: %v", err, err)
+				}
+				if he.Status != tc.status || he.Message != tc.message {
+					t.Errorf("HTTPError = %+v, want status %d and message %q", he, tc.status, tc.message)
+				}
+			})
+		}
+	}
+}

internal/commands/scheduler.go

@@ -1,13 +1,12 @@
-// scheduler subcommand: show the state of Arc's background schedulers
-// (continuous queries and retention). On Arc OSS neither scheduler runs
-// (Enterprise feature); this command is the quickest way to explain why
-// a new CQ or policy never fires.
+// scheduler subcommand: inspect and administer Arc's background schedulers
+// for continuous queries and retention.
 package commands
 
 import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"github.com/spf13/cobra"
 
@@ -20,7 +19,116 @@ func newSchedulerCmd() *cobra.Command {
 		Use:   "scheduler",
 		Short: "Inspect the continuous-query and retention schedulers",
 	}
-	c.AddCommand(newSchedulerStatusCmd())
+	c.AddCommand(newSchedulerStatusCmd(), newSchedulerCQCmd(), newSchedulerRetentionCmd())
+	return c
+}
+
+func newSchedulerCQCmd() *cobra.Command {
+	c := &cobra.Command{
+		Use:   "cq",
+		Short: "Administer the continuous-query scheduler",
+	}
+	c.AddCommand(newSchedulerCQReloadCmd())
+	return c
+}
+
+func newSchedulerCQReloadCmd() *cobra.Command {
+	var (
+		f            connFlags
+		outputFormat string
+	)
+	c := &cobra.Command{
+		Use:   "reload",
+		Short: "Reload all active continuous queries into the scheduler (admin)",
+		Long: `Reload all active continuous queries into the scheduler
+(POST /api/v1/schedulers/cq/reload). This refreshes scheduled jobs; it does
+not execute a query. Use "arcli cq execute <id|name>" to run one query now.`,
+		Args: cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if !validObjectFormat(outputFormat) {
+				return fmt.Errorf("invalid --output %q (valid: table, json)", outputFormat)
+			}
+			cli, _, err := f.client(cmd)
+			if err != nil {
+				return err
+			}
+			ctx, cancel := f.ctx(cmd)
+			defer cancel()
+			res, err := cli.ReloadCQScheduler(ctx)
+			if err != nil {
+				return err
+			}
+			if outputFormat == output.FormatJSON {
+				return writeRawJSON(cmd.OutOrStdout(), res.Raw)
+			}
+			if _, err := fmt.Fprintln(cmd.OutOrStdout(), clean(res.Message)); err != nil {
+				return err
+			}
+			if res.JobCount != nil {
+				_, err := fmt.Fprintf(cmd.OutOrStdout(), "scheduled jobs: %d\n", *res.JobCount)
+				return err
+			}
+			return nil
+		},
+	}
+	f.addWithTimeout(c, 10*time.Minute)
+	c.Flags().StringVarP(&outputFormat, "output", "o", output.FormatTable, "output format: table|json")
+	return c
+}
+
+func newSchedulerRetentionCmd() *cobra.Command {
+	c := &cobra.Command{
+		Use:   "retention",
+		Short: "Administer the retention scheduler",
+	}
+	c.AddCommand(newSchedulerRetentionTriggerCmd())
+	return c
+}
+
+func newSchedulerRetentionTriggerCmd() *cobra.Command {
+	var (
+		f            connFlags
+		outputFormat string
+		yes          bool
+	)
+	c := &cobra.Command{
+		Use:   "trigger",
+		Short: "Run all active retention policies now (admin; may delete data)",
+		Long: `Run all active retention policies on this server immediately
+(POST /api/v1/schedulers/retention/trigger). Policies may permanently delete
+old data. This is different from "arcli retention execute <id|name>", which
+runs one selected policy. The server does not provide a dry run for this
+scheduler action, so arcli asks for confirmation; pass --yes for scripts.`,
+		Example: `  arcli scheduler retention trigger
+  arcli scheduler retention trigger --yes -o json`,
+		Args: cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			if !validObjectFormat(outputFormat) {
+				return fmt.Errorf("invalid --output %q (valid: table, json)", outputFormat)
+			}
+			cli, _, err := f.client(cmd)
+			if err != nil {
+				return err
+			}
+			if err := confirmOrAbort(cmd, "Run every active retention policy now? This may permanently delete data.", yes); err != nil {
+				return err
+			}
+			ctx, cancel := f.ctx(cmd)
+			defer cancel()
+			res, err := cli.TriggerRetentionScheduler(ctx)
+			if err != nil {
+				return err
+			}
+			if outputFormat == output.FormatJSON {
+				return writeRawJSON(cmd.OutOrStdout(), res.Raw)
+			}
+			_, err = fmt.Fprintln(cmd.OutOrStdout(), clean(res.Message))
+			return err
+		},
+	}
+	f.addWithTimeout(c, 30*time.Minute)
+	c.Flags().StringVarP(&outputFormat, "output", "o", output.FormatTable, "output format: table|json")
+	c.Flags().BoolVarP(&yes, "yes", "y", false, "skip the confirmation prompt")
 	return c
 }