test(integration): restore Go-native PostgreSQL MySQL and CSV coverage

.github/workflows/integration.yml

@@ -7,6 +7,8 @@ on:
 jobs:
   integration:
     runs-on: ubuntu-latest
+    permissions:
+      contents: read
     services:
       postgres:
         image: postgres:17
@@ -22,6 +24,21 @@ jobs:
           --health-timeout 5s
           --health-retries 5
 
+      mysql:
+        image: mysql:8.0
+        env:
+          MYSQL_ROOT_PASSWORD: rootpassword
+          MYSQL_USER: myuser
+          MYSQL_PASSWORD: mypassword
+          MYSQL_DATABASE: mydb
+        ports:
+          - 3306:3306
+        options: >-
+          --health-cmd "mysqladmin ping -h 127.0.0.1 -uroot -prootpassword --silent"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 10
+
     steps:
       - name: Checkout code
         uses: actions/checkout@v7
@@ -69,6 +86,13 @@ jobs:
       - name: Run workload branch tests (Postgres)
         run: make run-workload-branches
 
+      - name: Run tagged baseline integration suite
+        env:
+          STROPPY_TMPFS_URL: postgres://postgres:postgres@127.0.0.1:5432/postgres?sslmode=disable
+          STROPPY_PG_URL: postgres://postgres:postgres@127.0.0.1:5432/postgres?sslmode=disable
+          STROPPY_MYSQL_URL: myuser:mypassword@tcp(127.0.0.1:3306)/mydb?parseTime=true&multiStatements=true
+        run: make integration
+
       - name: Upload binary artifact
         uses: actions/upload-artifact@v7
         with:

AGENTS.md

@@ -22,6 +22,36 @@ make linter         # read-only check after linter_fix
 make tests          # all tests with race detector and coverage
 ```
 
+### Integration tests
+
+The mandatory tagged suite requires the built binary, Docker (for the OTEL
+collector), PostgreSQL, and MySQL. The local baseline harness uses PostgreSQL on
+`localhost:5434` and MySQL on `localhost:3307`:
+
+```bash
+make tmpfs-up
+make build
+make integration
+make tmpfs-down
+```
+
+`STROPPY_TMPFS_URL` / `STROPPY_PG_URL` and `STROPPY_MYSQL_URL` override those
+service URLs. Once selected, baseline tests fail when the binary, Docker daemon,
+or either database is unavailable; there is no integration skip switch.
+
+Picodata and YDB are a separate optional suite. Stop the baseline harness before
+starting the four-database harness because they share host ports:
+
+```bash
+make tmpfs-all-up
+make integration-optional
+make tmpfs-all-down
+```
+
+The large TPC-H SF=1 answer check is separate as `make integration-sf1` and uses
+the baseline PostgreSQL service. CI builds once, runs both smoke tiers and
+`make integration`, then uploads that same tested binary.
+
 Application configuration is plain Go under `pkg/config`; `config.Unmarshal` owns
 strict JSON/ProtoJSON compatibility. Regenerate the committed file-envelope schema
 with `go generate ./pkg/config`. There is no application protobuf generation step,
@@ -93,10 +123,10 @@ Nonfatal terminal transaction errors fail one iteration and the VU continues;
 query-set workloads count the failed query and continue. Both exit 0 but produce
 bounded WARN aggregates and a prominent final completed-with-errors summary.
 Scheduled retries are counted separately and do not mark a run failed when a
-later attempt succeeds. Setup, workload teardown, fatal, and cancellation retain
-their nonzero or signal-derived semantics; fatal stops the scenario and is
-reported once. TPC-H/TPC-DS SF=1 answer comparison remains diagnostic-only.
-The removed `errorMode` driver field and alias are rejected.
+later attempt succeeds. Setup, workload or driver teardown, fatal, and
+cancellation retain their nonzero or signal-derived semantics; fatal stops the
+scenario and is reported once. TPC-H/TPC-DS SF=1 answer comparison remains
+diagnostic-only. The removed `errorMode` driver field and alias are rejected.
 
 ## CLI Usage
 

CHANGELOG.md

@@ -46,6 +46,7 @@ Group lines under `Added` / `Changed` / `Fixed` / `Removed`. Append a PR link
 
 ### Fixed
 
+- CSV output now publishes shards, merged files, and manifests atomically across fresh and repeated loads, so canceled or failed loads retain recoverable shards and never expose partial output as complete. ([#157](https://github.com/stroppy-io/stroppy/pull/157))
 - Query helpers now return an empty result instead of panicking when a driver supplies no result set, and query timeouts that surface while closing result sets are reported once instead of repeating the same error. ([#153](https://github.com/stroppy-io/stroppy/pull/153))
 - Empty, whitespace-only, or comma-only step filters no longer conflict with a real opposite filter; `--steps=` still clears configured steps before `--no-steps` is applied. ([#149](https://github.com/stroppy-io/stroppy/pull/149))
 - The `workload` step is silent on the console again (no `Start`/`End` record per transaction), while setup/load/schema steps still log a single start/end and the `simple` workload now honors `--steps`/`--no-steps` like the other workloads. ([#149](https://github.com/stroppy-io/stroppy/pull/149))

Makefile

@@ -63,6 +63,20 @@ linter_fix: # Start linter with possible fixes (NEVER run casually — rewrites
 tests: # Run tests with coverage; pass TEST_FLAGS=-short for the CI-sized suite
 	go test -race $(TEST_FLAGS) ./... -coverprofile=coverage.out
 
+.PHONY: integration integration-optional integration-sf1
+
+integration: # Run mandatory tagged PostgreSQL, MySQL, CSV, and OTEL integration tests
+	@test -x $(STROPPY_OUT_FILE) || { echo "error: $(STROPPY_OUT_FILE) is missing; run 'make build' first"; exit 1; }
+	go test -tags=integration -count=1 -timeout=15m ./test/integration
+
+integration-optional: # Run optional Picodata and YDB tagged integration tests
+	@test -x $(STROPPY_OUT_FILE) || { echo "error: $(STROPPY_OUT_FILE) is missing; run 'make build' first"; exit 1; }
+	go test -tags='integration integration_optional' -count=1 -timeout=30m -run 'TestTpchLoadOn(Picodata|YDB)' ./test/integration
+
+integration-sf1: # Run the explicit heavy TPC-H SF=1 answer validation
+	@test -x $(STROPPY_OUT_FILE) || { echo "error: $(STROPPY_OUT_FILE) is missing; run 'make build' first"; exit 1; }
+	go test -tags='integration integration_sf1' -count=1 -timeout=75m -run TestTpchAnswersSpotCheck ./test/integration
+
 
 ##
 ## Reference-data JSON regeneration (build-time, run with upstream inputs)
@@ -185,30 +199,30 @@ run-workload-branches: # Tier 1: real-Postgres smoke of both branches (tpcb/tpcc
 	exit $$rc
 
 ##
-## Tmpfs Postgres integration harness
+## Baseline PostgreSQL + MySQL integration harness
 ##
 
 .PHONY: tmpfs-up tmpfs-down tmpfs-clean tmpfs-psql
 
-tmpfs-up: # Start tmpfs Postgres container for integration tests
+tmpfs-up: # Start baseline tmpfs PostgreSQL and MySQL services
 	docker compose -f test/compose.tmpfs.yml up -d --wait
 
-tmpfs-down: # Stop and remove tmpfs Postgres container and volumes
+tmpfs-down: # Stop baseline integration services and remove their volumes
 	docker compose -f test/compose.tmpfs.yml down -v
 
-tmpfs-clean: # Recycle the tmpfs Postgres container; discards all data
+tmpfs-clean: # Recycle baseline integration services; discard all data
 	$(MAKE) tmpfs-down && $(MAKE) tmpfs-up
 
 tmpfs-psql: # Open psql shell into the tmpfs Postgres container
 	docker exec -it stroppy-pg-tmpfs psql -U postgres -d stroppy
 
 ##
-## Multi-DB tmpfs integration harness (postgres + mysql + picodata + ydb)
+## Optional multi-DB tmpfs integration harness
 ##
 
 .PHONY: tmpfs-all-up tmpfs-all-down tmpfs-all-clean
 
-tmpfs-all-up: # Start all 4 DBs (pg, mysql, picodata, ydb) on non-default ports
+tmpfs-all-up: # Start optional PostgreSQL, MySQL, Picodata, and YDB harness
 	docker compose -f test/compose.tmpfs-all.yml up -d --wait pg-tmpfs-all mysql-tmpfs-all picodata-tmpfs-all ydb-tmpfs-all
 	docker compose -f test/compose.tmpfs-all.yml up picodata-init
 

cmd/stroppy/commands/help/topic_drivers.go

@@ -137,13 +137,13 @@ ERROR AND EXIT BEHAVIOR
   iterations, failed queries, retries, and representative groups. Nonfatal
   errors leave the process exit status at 0.
 
-  Setup failures, workload teardown failures, and fatal workload actions return
-  a nonzero status. TPC-H and TPC-DS SF=1 answer comparison remains diagnostic:
-  query errors and answer differences are logged but do not change exit status.
-  A fatal action stops the whole scenario and is reported once. SIGINT/SIGTERM
-  cancellation stops the scenario, runs teardown, and keeps its signal-derived
-  exit status (130/143; a forced second signal is 2). Cancellation is not
-  reported as a nonfatal benchmark error.
+  Setup failures, workload or driver teardown failures, and fatal workload
+  actions return a nonzero status. TPC-H and TPC-DS SF=1 answer comparison
+  remains diagnostic: query errors and answer differences are logged but do not
+  change exit status. A fatal action stops the whole scenario and is reported
+  once. SIGINT/SIGTERM cancellation stops the scenario, runs teardown, and keeps
+  its signal-derived exit status (130/143; a forced second signal is 2).
+  Cancellation is not reported as a nonfatal benchmark error.
 
   To inspect the driver insert methods each driver supports:
 

pkg/bench/runtime.go

@@ -244,6 +244,15 @@ func Run(
 		return fmt.Errorf("driver dispatch: %w", err)
 	}
 
+	defer func() {
+		teardownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), teardownTimeout)
+		defer cancel()
+
+		if err := drv.Teardown(teardownCtx); err != nil {
+			retErr = errors.Join(retErr, fmt.Errorf("driver teardown: %w", err))
+		}
+	}()
+
 	setupVU := &VU{root: root, vuid: 1, initPhase: true, ctx: ctx}
 	setupBench := &Bench{
 		root: root, vu: setupVU,

pkg/bench/runtime_test.go

@@ -14,7 +14,9 @@ import (
 	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/stroppy-io/stroppy/pkg/config"
+	"github.com/stroppy-io/stroppy/pkg/driver"
 	_ "github.com/stroppy-io/stroppy/pkg/driver/noop"
+	"github.com/stroppy-io/stroppy/pkg/driver/stats"
 )
 
 func TestRunScenarioReturnsFatalErrorAndCancelsWorkers(t *testing.T) {
@@ -243,6 +245,169 @@ func (w *fatalContextWorkload) Iterate(ctx context.Context, _ *Bench) error {
 
 func (*fatalContextWorkload) Teardown(context.Context, *Bench) error { return nil }
 
+const teardownLifecycleDriverType config.DriverType = 1000
+
+type driverTeardownContextKey struct{}
+
+var (
+	registerTeardownLifecycleOnce sync.Once
+	teardownLifecycleDriverRun    *teardownLifecycleDriver
+	teardownLifecycleWorkloadRun  *teardownLifecycleWorkload
+	errTeardownLifecycleBegin     = errors.New("test driver does not support transactions")
+)
+
+func registerTeardownLifecycleTest() {
+	registerTeardownLifecycleOnce.Do(func() {
+		driver.RegisterDriver(teardownLifecycleDriverType, func(context.Context, driver.Options) (driver.Driver, error) {
+			if teardownLifecycleDriverRun == nil {
+				return &teardownLifecycleDriver{}, nil
+			}
+
+			return teardownLifecycleDriverRun, nil
+		})
+		Register(func() Workload {
+			if teardownLifecycleWorkloadRun == nil {
+				return &teardownLifecycleWorkload{}
+			}
+
+			return teardownLifecycleWorkloadRun
+		})
+	})
+}
+
+func TestRunFinalizesDriverAfterWorkload(t *testing.T) {
+	registerTeardownLifecycleTest()
+
+	setupErr := errors.New("setup sentinel")
+	workloadErr := errors.New("workload teardown sentinel")
+	driverErr := errors.New("driver teardown sentinel")
+	recorder := &teardownLifecycleRecorder{}
+
+	teardownLifecycleWorkloadRun = &teardownLifecycleWorkload{
+		recorder:    recorder,
+		setupErr:    setupErr,
+		teardownErr: workloadErr,
+	}
+	teardownLifecycleDriverRun = &teardownLifecycleDriver{
+		recorder:    recorder,
+		teardownErr: driverErr,
+	}
+
+	ctx, cancel := context.WithCancel(context.WithValue(context.Background(), driverTeardownContextKey{}, "retained"))
+	cancel()
+
+	err := Run(
+		ctx,
+		"test/driver-teardown-lifecycle",
+		map[int]*config.DriverConfig{0: {DriverType: teardownLifecycleDriverType}},
+		ParamInputs{},
+		nil,
+		nil,
+		zap.NewNop(),
+		&MetricsConfig{},
+	)
+	for _, want := range []error{setupErr, workloadErr, driverErr} {
+		if !errors.Is(err, want) {
+			t.Errorf("Run() error = %v, want joined %v", err, want)
+		}
+	}
+
+	if got := strings.Join(recorder.order, ","); got != "workload,driver" {
+		t.Fatalf("teardown order = %q, want workload,driver", got)
+	}
+
+	if recorder.workloadCalls != 1 || recorder.driverCalls != 1 {
+		t.Fatalf("teardown calls = workload:%d driver:%d, want one each", recorder.workloadCalls, recorder.driverCalls)
+	}
+
+	if !recorder.workloadDetached || !recorder.driverDetached {
+		t.Fatalf("detached contexts = workload:%t driver:%t, want true", recorder.workloadDetached, recorder.driverDetached)
+	}
+
+	if !recorder.workloadDeadline || !recorder.driverDeadline {
+		t.Fatalf("deadline contexts = workload:%t driver:%t, want true", recorder.workloadDeadline, recorder.driverDeadline)
+	}
+
+	if teardownLifecycleWorkloadRun.contextValue != "retained" || teardownLifecycleDriverRun.contextValue != "retained" {
+		t.Fatalf(
+			"teardown context values = workload:%v driver:%v, want retained",
+			teardownLifecycleWorkloadRun.contextValue,
+			teardownLifecycleDriverRun.contextValue,
+		)
+	}
+}
+
+type teardownLifecycleRecorder struct {
+	order            []string
+	workloadCalls    int
+	driverCalls      int
+	workloadDetached bool
+	driverDetached   bool
+	workloadDeadline bool
+	driverDeadline   bool
+}
+
+type teardownLifecycleWorkload struct {
+	recorder     *teardownLifecycleRecorder
+	setupErr     error
+	teardownErr  error
+	contextValue any
+}
+
+func (*teardownLifecycleWorkload) Name() string                          { return "test/driver-teardown-lifecycle" }
+func (*teardownLifecycleWorkload) Define(*Def) error                     { return nil }
+func (w *teardownLifecycleWorkload) Setup(context.Context, *Bench) error { return w.setupErr }
+func (*teardownLifecycleWorkload) Iterate(context.Context, *Bench) error { return nil }
+func (w *teardownLifecycleWorkload) Teardown(ctx context.Context, _ *Bench) error {
+	if w.recorder == nil {
+		return w.teardownErr
+	}
+
+	w.recorder.order = append(w.recorder.order, "workload")
+	w.recorder.workloadCalls++
+	w.recorder.workloadDetached = ctx.Err() == nil
+	_, w.recorder.workloadDeadline = ctx.Deadline()
+	w.contextValue = ctx.Value(driverTeardownContextKey{})
+
+	return w.teardownErr
+}
+
+type teardownLifecycleDriver struct {
+	recorder     *teardownLifecycleRecorder
+	teardownErr  error
+	contextValue any
+}
+
+func (*teardownLifecycleDriver) Insert(context.Context, *driver.InsertRequest) (*stats.Query, error) {
+	return &stats.Query{}, nil
+}
+
+func (*teardownLifecycleDriver) RunQuery(context.Context, string, map[string]any) (*driver.QueryResult, error) {
+	return &driver.QueryResult{}, nil
+}
+
+func (*teardownLifecycleDriver) Begin(context.Context, config.TxIsolationLevel) (driver.Tx, error) {
+	return nil, errTeardownLifecycleBegin
+}
+
+func (*teardownLifecycleDriver) ClassifyError(err error) driver.ErrorFacts {
+	return driver.DefaultErrorFacts(err)
+}
+
+func (d *teardownLifecycleDriver) Teardown(ctx context.Context) error {
+	if d.recorder == nil {
+		return d.teardownErr
+	}
+
+	d.recorder.order = append(d.recorder.order, "driver")
+	d.recorder.driverCalls++
+	d.recorder.driverDetached = ctx.Err() == nil
+	_, d.recorder.driverDeadline = ctx.Deadline()
+	d.contextValue = ctx.Value(driverTeardownContextKey{})
+
+	return d.teardownErr
+}
+
 func installRuntimeTestRoot(t *testing.T) {
 	t.Helper()