fix: propagate manifest list errors, keep all-null columns, real fdatasync (#314, #337, #305)

RELEASE_NOTES_2026.09.1.md

@@ -57,6 +57,28 @@ Reachable only when RBAC is enabled (the multi-tenant authorization boundary); n
 
 ## Bug fixes
 
+### Compaction no longer treats a storage failure as "no manifests exist" ([#314](https://github.com/Basekick-Labs/arc/issues/314))
+
+`ListManifests` discarded every error from the storage backend and returned an empty list, with no log line. "No manifests exist" and "we could not find out" are opposite instructions to the caller: an empty manifest set means nothing is being compacted and the candidate may proceed, while a failed lookup means the files in flight cannot be identified.
+
+The candidate filter already had the correct guard — it skips the partition when the manifest lookup fails, explicitly *"to avoid re-compaction"* — but swallowing the error inside `ListManifests` made that guard unreachable. A transient failure (S3 throttling, an expired credential, a network blip) therefore let Arc treat files already claimed by an in-flight compaction as untracked, and compact them a second time.
+
+Errors now propagate. A missing manifest directory remains a non-error: the local backend skips directories that do not exist, and the object-store backends return an empty listing for a prefix with no objects, so any error reaching this layer is a real failure.
+
+### Ingest no longer drops columns that are entirely null in a batch ([#337](https://github.com/Basekick-Labs/arc/issues/337))
+
+A column whose every value was null in a batch was dropped, because no type can be inferred from it. In most cases readers absorbed this — queries and compaction union schemas by name — but a column that is null in **every** batch never appeared in any file, so querying it failed with `Binder Error: Referenced column "depth" not found` instead of returning NULLs. Realistic triggers: an optional field absent for a whole batch, or a sensor reporting null through an outage.
+
+Such columns are now written as an all-null placeholder that keeps the column present and every value NULL. A later batch carrying real values still infers its own type, so nothing is pinned to the placeholder. The `time` column is exempt and an all-null time is now rejected outright: a VARCHAR time column makes a partition un-compactable.
+
+### `fdatasync` WAL sync mode is now actually fdatasync on Linux ([#305](https://github.com/Basekick-Labs/arc/issues/305))
+
+`wal.sync_mode` accepted `fdatasync`, and it is the default when the WAL is enabled, but both `fsync` and `fdatasync` called the same full `Sync()`. Operators who selected the balanced mode silently got the strictest one.
+
+Linux now uses a real `fdatasync(2)`, which skips the metadata-journal flush (retried on `EINTR` so an interrupted call cannot report durability it did not achieve). Go does not expose `fdatasync` on macOS or Windows, so those platforms honestly fall back to a full `Sync()` and now **log that fact once at startup** rather than reporting a mode they are not performing. The startup log also carries `fdatasync_supported` so the effective behavior is visible.
+
+Expect no measurable throughput change: WAL syncs are driven by a 100 ms ticker rather than per-write, so there are at most ~10 per second regardless of ingest rate. This is a correctness and honesty fix, not a performance one.
+
 ### Features share the auth manager's SQLite handle instead of opening their own ([#329](https://github.com/Basekick-Labs/arc/issues/329), [#562](https://github.com/Basekick-Labs/arc/issues/562))
 
 Arc keeps auth, audit, tiering, governance, retention, continuous-query and MQTT metadata in a single SQLite file (`auth.db_path`, default `./data/arc.db`). Each of those features opened its **own** connection to that file, so a default deployment ran six independent connection pools — each capped at one connection, since SQLite has a single writer — competing for the same write lock. They now borrow the auth manager's existing handle instead.

internal/compaction/manifest.go

@@ -142,12 +142,25 @@ func (m *ManifestManager) ReadManifest(ctx context.Context, manifestPath string)
 	return &manifest, nil
 }
 
-// ListManifests lists all manifest files in storage
+// ListManifests lists all manifest files in storage.
+//
+// Errors are returned rather than reported as an empty list. "No manifests
+// exist" and "storage is unreachable" mean opposite things to the caller:
+// filterCandidateFiles treats an empty manifest set as "nothing is being
+// compacted, proceed" but has an explicit guard that skips the partition when
+// the lookup fails. Collapsing the error into an empty slice made that guard
+// unreachable, so a transient List failure (S3 throttling, expired
+// credentials, a network blip) let Arc re-compact files another job already
+// had in flight.
+//
+// A missing manifest directory is not an error at this layer: LocalBackend.List
+// skips directories that do not exist, and the object-store backends return an
+// empty result for a prefix with no objects. So any error arriving here is a
+// real failure.
 func (m *ManifestManager) ListManifests(ctx context.Context) ([]string, error) {
 	objects, err := m.backend.List(ctx, ManifestBasePath+"/")
 	if err != nil {
-		// If the directory doesn't exist, return empty list
-		return []string{}, nil
+		return nil, fmt.Errorf("failed to list manifests: %w", err)
 	}
 
 	var manifests []string
@@ -165,7 +178,8 @@ func (m *ManifestManager) ListManifests(ctx context.Context) ([]string, error) {
 func (m *ManifestManager) RecoverOrphanedManifests(ctx context.Context) (int, error) {
 	manifests, err := m.ListManifests(ctx)
 	if err != nil {
-		return 0, fmt.Errorf("failed to list manifests: %w", err)
+		// ListManifests already describes the failure.
+		return 0, err
 	}
 
 	if len(manifests) == 0 {

internal/compaction/manifest_errors_test.go

@@ -0,0 +1,102 @@
+package compaction
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/basekick-labs/arc/internal/storage"
+	"github.com/rs/zerolog"
+)
+
+// listFailingBackend fails every List call, standing in for S3 throttling,
+// expired credentials, or a network blip.
+type listFailingBackend struct {
+	storage.Backend
+	err error
+}
+
+func (b *listFailingBackend) List(ctx context.Context, prefix string) ([]string, error) {
+	return nil, b.err
+}
+
+func newListFailingManager(t *testing.T, err error) *ManifestManager {
+	t.Helper()
+	local, lerr := storage.NewLocalBackend(t.TempDir(), zerolog.Nop())
+	if lerr != nil {
+		t.Fatalf("NewLocalBackend: %v", lerr)
+	}
+	return NewManifestManager(&listFailingBackend{Backend: local, err: err}, zerolog.Nop())
+}
+
+// A storage failure must surface as an error, not as "there are no manifests".
+//
+// The two are opposite instructions to the caller: an empty manifest set means
+// "nothing is being compacted, proceed", while a failed lookup means "we cannot
+// tell, so do not touch these files".
+func TestListManifests_PropagatesStorageErrors(t *testing.T) {
+	sentinel := errors.New("SlowDown: please reduce your request rate")
+	m := newListFailingManager(t, sentinel)
+
+	manifests, err := m.ListManifests(context.Background())
+	if err == nil {
+		t.Fatalf("expected the storage error to propagate; got nil with %d manifests", len(manifests))
+	}
+	if !errors.Is(err, sentinel) {
+		t.Errorf("underlying storage error should be wrapped, got: %v", err)
+	}
+	if !strings.Contains(err.Error(), "list manifests") {
+		t.Errorf("error should name the operation, got: %v", err)
+	}
+}
+
+// GetFilesInManifests feeds filterCandidateFiles, which has an explicit guard
+// that skips a partition when this lookup fails. Swallowing the error made that
+// guard unreachable and allowed re-compaction of in-flight files.
+func TestGetFilesInManifests_PropagatesStorageErrors(t *testing.T) {
+	m := newListFailingManager(t, errors.New("connection reset by peer"))
+
+	files, err := m.GetFilesInManifests(context.Background())
+	if err == nil {
+		t.Fatalf("expected the storage error to propagate; got nil with %d files — "+
+			"filterCandidateFiles would read this as 'nothing is being compacted' and proceed", len(files))
+	}
+}
+
+// Recovery must not report success over a failed listing: returning (0, nil)
+// says "there was nothing to recover", which is a different claim from "we
+// could not find out".
+func TestRecoverOrphanedManifests_PropagatesStorageErrors(t *testing.T) {
+	m := newListFailingManager(t, errors.New("AccessDenied"))
+
+	recovered, err := m.RecoverOrphanedManifests(context.Background())
+	if err == nil {
+		t.Fatalf("expected the storage error to propagate; got nil with recovered=%d", recovered)
+	}
+	if recovered != 0 {
+		t.Errorf("recovered = %d, want 0 on failure", recovered)
+	}
+	// The message should not stutter now that ListManifests describes the failure.
+	if strings.Count(err.Error(), "failed to list manifests") > 1 {
+		t.Errorf("error message is double-wrapped: %v", err)
+	}
+}
+
+// A missing manifest directory is normal on a fresh install and must stay a
+// non-error: the backends already report an absent prefix as an empty listing.
+func TestListManifests_EmptyStorageIsNotAnError(t *testing.T) {
+	local, err := storage.NewLocalBackend(t.TempDir(), zerolog.Nop())
+	if err != nil {
+		t.Fatalf("NewLocalBackend: %v", err)
+	}
+	m := NewManifestManager(local, zerolog.Nop())
+
+	manifests, err := m.ListManifests(context.Background())
+	if err != nil {
+		t.Fatalf("a fresh install with no manifest directory must not error: %v", err)
+	}
+	if len(manifests) != 0 {
+		t.Errorf("expected no manifests, got %d", len(manifests))
+	}
+}

internal/ingest/allnil_columns_test.go

@@ -0,0 +1,146 @@
+package ingest
+
+import (
+	"strings"
+	"testing"
+)
+
+// An all-nil column used to be dropped, because no type can be inferred from
+// it. That made the column absent from the batch's parquet file; a column that
+// is all-nil in every batch never appeared in any file, so querying it failed
+// to bind instead of returning NULLs (#337).
+func TestConvertColumnsToTyped_AllNilColumnIsPreserved(t *testing.T) {
+	buffer := createTestArrowBuffer(t)
+
+	batch, n, err := buffer.convertColumnsToTyped("sensors", map[string][]interface{}{
+		"time":  {int64(1609459200000000), int64(1609459200000001), int64(1609459200000002)},
+		"value": {1.0, 2.0, 3.0},
+		"depth": {nil, nil, nil},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if n != 3 {
+		t.Fatalf("numRecords = %d, want 3", n)
+	}
+
+	col, ok := batch.Data["depth"]
+	if !ok {
+		t.Fatal("all-nil column was dropped — it will be missing from the parquet file")
+	}
+
+	arr, ok := col.([]string)
+	if !ok {
+		t.Fatalf("all-nil column type = %T, want []string placeholder", col)
+	}
+	if len(arr) != 3 {
+		t.Errorf("all-nil column length = %d, want 3 (must match the record count)", len(arr))
+	}
+
+	// Every entry must be marked NULL, not empty-string.
+	valid, ok := batch.Validity["depth"]
+	if !ok {
+		t.Fatal("all-nil column has no validity mask — entries would be written as empty strings, not NULL")
+	}
+	if len(valid) != 3 {
+		t.Fatalf("validity length = %d, want 3", len(valid))
+	}
+	for i, v := range valid {
+		if v {
+			t.Errorf("validity[%d] = true, want false — every value in the column is nil", i)
+		}
+	}
+}
+
+// The column must still line up with the rest of the batch, otherwise the
+// Arrow record builder produces a ragged batch.
+func TestConvertColumnsToTyped_AllNilColumnMatchesRecordCount(t *testing.T) {
+	buffer := createTestArrowBuffer(t)
+
+	const rows = 5
+	times := make([]interface{}, rows)
+	values := make([]interface{}, rows)
+	nils := make([]interface{}, rows)
+	for i := 0; i < rows; i++ {
+		times[i] = int64(1609459200000000 + i)
+		values[i] = float64(i)
+	}
+
+	batch, n, err := buffer.convertColumnsToTyped("m", map[string][]interface{}{
+		"time":    times,
+		"value":   values,
+		"missing": nils,
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if n != rows {
+		t.Fatalf("numRecords = %d, want %d", n, rows)
+	}
+	if got := len(batch.Data["missing"].([]string)); got != rows {
+		t.Errorf("all-nil column length = %d, want %d", got, rows)
+	}
+}
+
+// A later batch carrying real values must infer its own type — the string
+// placeholder applies only to the batch where the column was entirely nil.
+func TestConvertColumnsToTyped_AllNilDoesNotPinTypeForLaterBatches(t *testing.T) {
+	buffer := createTestArrowBuffer(t)
+
+	if _, _, err := buffer.convertColumnsToTyped("m", map[string][]interface{}{
+		"time":  {int64(1609459200000000)},
+		"depth": {nil},
+	}); err != nil {
+		t.Fatalf("first batch: %v", err)
+	}
+
+	batch, _, err := buffer.convertColumnsToTyped("m", map[string][]interface{}{
+		"time":  {int64(1609459200000001)},
+		"depth": {12.5},
+	})
+	if err != nil {
+		t.Fatalf("second batch: %v", err)
+	}
+	if _, ok := batch.Data["depth"].([]float64); !ok {
+		t.Errorf("depth type = %T, want []float64 — a real value must infer its own type", batch.Data["depth"])
+	}
+}
+
+// A partially-nil column already inferred its type from the non-nil values;
+// that behavior must be unchanged.
+func TestConvertColumnsToTyped_PartiallyNilColumnUnchanged(t *testing.T) {
+	buffer := createTestArrowBuffer(t)
+
+	batch, _, err := buffer.convertColumnsToTyped("m", map[string][]interface{}{
+		"time":  {int64(1609459200000000), int64(1609459200000001)},
+		"depth": {nil, 12.5},
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if _, ok := batch.Data["depth"].([]float64); !ok {
+		t.Errorf("depth type = %T, want []float64", batch.Data["depth"])
+	}
+	valid := batch.Validity["depth"]
+	if len(valid) != 2 || valid[0] || !valid[1] {
+		t.Errorf("validity = %v, want [false true]", valid)
+	}
+}
+
+// Time is exempt: a VARCHAR time column makes the partition un-compactable
+// (TIMESTAMP != VARCHAR bind failure), so an all-nil time is rejected rather
+// than written as a string placeholder.
+func TestConvertColumnsToTyped_AllNilTimeIsRejected(t *testing.T) {
+	buffer := createTestArrowBuffer(t)
+
+	_, _, err := buffer.convertColumnsToTyped("m", map[string][]interface{}{
+		"time":  {nil, nil},
+		"value": {1.0, 2.0},
+	})
+	if err == nil {
+		t.Fatal("expected an all-nil time column to be rejected")
+	}
+	if !strings.Contains(err.Error(), "time column") {
+		t.Errorf("error should name the time column, got: %v", err)
+	}
+}

internal/ingest/arrow_writer.go

@@ -1926,7 +1926,34 @@ func (b *ArrowBuffer) convertColumnsToTyped(measurement string, columns map[stri
 		// Infer type from first non-nil value
 		firstVal := firstNonNil(col)
 		if firstVal == nil {
-			continue // Skip all-nil columns
+			// Every value is nil, so there is nothing to infer a type from.
+			// Dropping the column would make it vanish from this batch's
+			// parquet file; a column that is all-nil in every batch would then
+			// never exist at all, and querying it fails to bind instead of
+			// returning NULLs (#337).
+			//
+			// Emit it as an all-null string column instead. The value is
+			// correct either way — every entry is NULL — and string is the
+			// safe placeholder: a later batch carrying real values writes its
+			// own inferred type, and readers union across files by name
+			// (read_parquet union_by_name=true), so the type only has to be
+			// consistent within a file, not across them.
+			//
+			// Time is exempt: a partition whose time column is VARCHAR cannot be
+			// compacted (TIMESTAMP != VARCHAR bind failure), so an all-nil time
+			// is rejected here rather than written as a string. The msgpack path
+			// already rejects it upstream in normalizeTimestamps, but this
+			// function is the chokepoint every typed write passes through, so
+			// the guard belongs here too.
+			if name == "time" {
+				return nil, 0, fmt.Errorf("time column contains only null values in measurement '%s' (writer must send an integer/float timestamp)", measurement)
+			}
+
+			arr := make([]string, len(col))
+			valid := make([]bool, len(col)) // all false — every entry is NULL
+			typed[name] = arr
+			validity[name] = valid
+			continue
 		}
 
 		// The "time" column is always int64 microseconds → Arrow Timestamp.

internal/wal/datasync_linux.go

@@ -0,0 +1,31 @@
+//go:build linux
+
+package wal
+
+import (
+	"os"
+	"syscall"
+)
+
+// dataSyncSupported reports whether dataSync performs a real fdatasync(2)
+// rather than falling back to a full fsync.
+const dataSyncSupported = true
+
+// dataSync flushes file data without waiting for a metadata-only journal
+// commit, using fdatasync(2).
+//
+// Note this is not free of metadata work: the WAL file grows on every append,
+// and fdatasync must still persist a changed file size before returning. The
+// saving is the inode's other metadata (timestamps), so the win over fsync is
+// real but modest — larger on rotational and network-backed storage than on
+// NVMe.
+func dataSync(f *os.File) error {
+	// Repeat on EINTR: a signal can interrupt fdatasync before it completes,
+	// and returning early would report durability that was not achieved.
+	for {
+		err := syscall.Fdatasync(int(f.Fd()))
+		if err != syscall.EINTR {
+			return err
+		}
+	}
+}