fix(adbc): re-authenticate on Unauthenticated instead of failing permanently

adbc.go

@@ -2,6 +2,7 @@ package gospice
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
 	"strings"
@@ -129,6 +130,37 @@ func (c *SpiceClient) SqlWithParams(ctx context.Context, sql string, params ...a
 		}
 	}
 
+	// Record the connection we are about to use so that, if it turns out to be
+	// stale, we only re-open it once even when many goroutines hit the failure
+	// at the same time.
+	used := c.adbcClient
+
+	rdr, err := c.execADBCWithBackoff(ctx, sql, params...)
+	if err != nil && isADBCAuthError(err) {
+		// The ADBC connection authenticates only once, when it is opened: a
+		// Basic-auth handshake yields a server-side session token that is then
+		// reused for every prepared statement on that connection. That session
+		// can be invalidated server-side (e.g. expired after a period of
+		// inactivity), after which the cached token is rejected on every
+		// subsequent request and the connection cannot recover on its own.
+		// Re-open the connection to perform a fresh handshake, then retry once.
+		if reinitErr := c.reinitADBC(used); reinitErr != nil {
+			return nil, fmt.Errorf("ADBC re-authentication failed: %w (original error: %v)", reinitErr, err)
+		}
+		rdr, err = c.execADBCWithBackoff(ctx, sql, params...)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return rdr, nil
+}
+
+// execADBCWithBackoff runs a parameterized ADBC query, retrying transient
+// (e.g. Unavailable / Internal) failures with the client's backoff policy.
+// Authentication failures are returned as-is (not retried here) so the caller
+// can re-establish the connection before retrying.
+func (c *SpiceClient) execADBCWithBackoff(ctx context.Context, sql string, params ...any) (array.RecordReader, error) {
 	var rdr array.RecordReader
 	err := backoff.Retry(func() error {
 		var err error
@@ -158,6 +190,45 @@ func (c *SpiceClient) SqlWithParams(ctx context.Context, sql string, params ...a
 	return rdr, nil
 }
 
+// isADBCAuthError reports whether err indicates the ADBC connection's
+// credentials/session were rejected by the server (as opposed to a transient
+// or query error). Such failures are not recoverable on the existing
+// connection and require re-opening it to perform a fresh handshake.
+func isADBCAuthError(err error) bool {
+	if err == nil {
+		return false
+	}
+	var adbcErr adbc.Error
+	if errors.As(err, &adbcErr) {
+		if adbcErr.Code == adbc.StatusUnauthenticated || adbcErr.Code == adbc.StatusUnauthorized {
+			return true
+		}
+	}
+	// Fall back to matching the message in case the typed error is not
+	// propagated through the wrapping chain.
+	msg := err.Error()
+	return strings.Contains(msg, "Unauthenticated") || strings.Contains(msg, "Invalid credentials")
+}
+
+// reinitADBC closes and re-opens the ADBC connection so that the next query
+// performs a fresh authentication handshake. The stale argument is the
+// connection the caller observed failing; if another goroutine has already
+// replaced it, this is a no-op so the connection is only re-opened once.
+func (c *SpiceClient) reinitADBC(stale *ADBCClient) error {
+	c.adbcMu.Lock()
+	defer c.adbcMu.Unlock()
+
+	// Another caller may have already re-opened the connection we observed as
+	// stale; if so, reuse theirs rather than churning the connection again.
+	if c.adbcClient != stale {
+		return nil
+	}
+
+	_ = c.closeADBC()
+	c.adbcClient = nil
+	return c.initADBC()
+}
+
 // QueryWithParams is deprecated. Use SqlWithParams instead.
 // Kept for backward compatibility with v7.
 func (c *SpiceClient) QueryWithParams(ctx context.Context, sql string, params ...any) (array.RecordReader, error) {

adbc_reauth_test.go

@@ -0,0 +1,62 @@
+package gospice
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/apache/arrow-adbc/go/adbc"
+)
+
+func TestIsADBCAuthError(t *testing.T) {
+	tests := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{
+			name: "nil",
+			err:  nil,
+			want: false,
+		},
+		{
+			name: "adbc unauthenticated",
+			err:  adbc.Error{Code: adbc.StatusUnauthenticated, Msg: "[FlightSQL] Invalid credentials"},
+			want: true,
+		},
+		{
+			name: "adbc unauthorized",
+			err:  adbc.Error{Code: adbc.StatusUnauthorized, Msg: "[FlightSQL] forbidden"},
+			want: true,
+		},
+		{
+			name: "wrapped adbc unauthenticated (the prepared-statement failure shape)",
+			err: fmt.Errorf("error preparing statement: %w",
+				adbc.Error{Code: adbc.StatusUnauthenticated, Msg: "[FlightSQL] Invalid credentials (Unauthenticated; Prepare)"}),
+			want: true,
+		},
+		{
+			name: "adbc internal is not an auth error",
+			err:  adbc.Error{Code: adbc.StatusInternal, Msg: "boom"},
+			want: false,
+		},
+		{
+			name: "plain error mentioning Unauthenticated (fallback)",
+			err:  errors.New("rpc error: code = Unauthenticated desc = invalid"),
+			want: true,
+		},
+		{
+			name: "unrelated transient error",
+			err:  errors.New("connection refused"),
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isADBCAuthError(tt.err); got != tt.want {
+				t.Errorf("isADBCAuthError(%v) = %v, want %v", tt.err, got, tt.want)
+			}
+		})
+	}
+}

client.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/apache/arrow-go/v18/arrow/flight"
@@ -40,6 +41,7 @@ type SpiceClient struct {
 
 	flightClient  flight.Client
 	adbcClient    *ADBCClient
+	adbcMu        sync.Mutex // guards re-initialization of adbcClient
 	httpClient    http.Client
 	backoffPolicy backoff.BackOff
 	maxRetries    uint