mirror of
https://github.com/t8y2/dbx.git
synced 2026-10-19 16:16:45 +00:00
fix(neo4j): fall back to direct connection on Bolt 3 routing errors
This commit is contained in:
@@ -12,6 +12,13 @@ This module replaces the Neo4j JDBC Agent with the official Neo4j Go Driver.
|
||||
|
||||
## Compatibility
|
||||
|
||||
- `neo4j://` (routing) is the default scheme. When a server reports that cluster
|
||||
routing is unavailable — Bolt 3 servers such as Neo4j 3.5, which never expose
|
||||
the routing procedure — the agent retries the same target with a direct
|
||||
`bolt://` connection. An explicit `scheme` parameter still wins, including the
|
||||
TLS variants (`bolt+s` / `bolt+ssc`).
|
||||
- Servers that negotiate Bolt 3 or older only expose the default database, so
|
||||
the agent does not send a database name for them (the driver rejects it).
|
||||
- Neo4j databases are discovered with `SHOW DATABASES`, with the configured database retained as a fallback for Community-compatible servers such as Memgraph.
|
||||
- Node labels use `CALL db.labels()`.
|
||||
- Properties use `db.schema.nodeTypeProperties()` with a sampled-node fallback.
|
||||
@@ -34,3 +41,7 @@ DBX_NEO4J_USER=neo4j \
|
||||
DBX_NEO4J_PASSWORD=password \
|
||||
go test -run TestLiveNeo4jAgent
|
||||
```
|
||||
|
||||
`TestLiveNeo4jAgentSingleInstanceRoutingFallback` uses the default routing
|
||||
scheme on purpose, so it exercises the direct-connection fallback; run it
|
||||
against a single-instance server that cannot route (for example Neo4j 3.5).
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
neo4j "github.com/neo4j/neo4j-go-driver/v6/neo4j"
|
||||
neo4jauth "github.com/neo4j/neo4j-go-driver/v6/neo4j/auth"
|
||||
"github.com/neo4j/neo4j-go-driver/v6/neo4j/config"
|
||||
neo4jdb "github.com/neo4j/neo4j-go-driver/v6/neo4j/db"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -27,6 +28,28 @@ const (
|
||||
)
|
||||
|
||||
func newConnectionRuntime(params connectParams) (*connectionRuntime, error) {
|
||||
runtime, err := openVerifiedRuntime(params)
|
||||
if err == nil {
|
||||
return runtime, nil
|
||||
}
|
||||
direct, changed := directConnectionParams(params)
|
||||
if !changed || !isRoutingUnsupported(err) {
|
||||
return nil, err
|
||||
}
|
||||
// Single-instance servers older than Neo4j 4 (which negotiate Bolt 3) do not
|
||||
// expose the cluster routing procedure, so `neo4j://` cannot be used against
|
||||
// them. Retry the same target with a direct `bolt://` connection instead of
|
||||
// failing the connection outright.
|
||||
runtime, directErr := openVerifiedRuntime(direct)
|
||||
if directErr != nil {
|
||||
return nil, directErr
|
||||
}
|
||||
return runtime, nil
|
||||
}
|
||||
|
||||
// openVerifiedRuntime opens a driver for params and fails unless the server
|
||||
// answers a connectivity check.
|
||||
func openVerifiedRuntime(params connectParams) (*connectionRuntime, error) {
|
||||
driver, err := openDriver(params)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -37,7 +60,19 @@ func newConnectionRuntime(params connectParams) (*connectionRuntime, error) {
|
||||
_ = driver.Close(context.Background())
|
||||
return nil, err
|
||||
}
|
||||
return &connectionRuntime{driver: driver, params: params}, nil
|
||||
return &connectionRuntime{driver: driver, params: params, legacySingleDatabase: usesLegacySingleDatabase(ctx, driver)}, nil
|
||||
}
|
||||
|
||||
// usesLegacySingleDatabase reports whether the server negotiated Bolt 3 or
|
||||
// older. Those servers only expose the default database, and the driver refuses
|
||||
// to select a database name for them, so sessions must not send one. Probe
|
||||
// failures keep the modern behavior.
|
||||
func usesLegacySingleDatabase(ctx context.Context, driver neo4j.Driver) bool {
|
||||
info, err := driver.GetServerInfo(ctx)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return info.ProtocolVersion().Major < 4
|
||||
}
|
||||
|
||||
func openDriver(params connectParams) (neo4j.Driver, error) {
|
||||
@@ -130,14 +165,93 @@ func buildNeo4jURI(params connectParams) (string, error) {
|
||||
if port <= 0 {
|
||||
port = defaultNeo4jPort
|
||||
}
|
||||
scheme := "neo4j"
|
||||
return (&url.URL{Scheme: effectiveNeo4jScheme(params), Host: net.JoinHostPort(host, strconv.Itoa(port))}).String(), nil
|
||||
}
|
||||
|
||||
// effectiveNeo4jScheme resolves the URI scheme a connection would use, in the
|
||||
// same order as buildNeo4jURI: an explicit connection string wins, then an
|
||||
// explicit `scheme` parameter, then TLS settings imply `neo4j+s`.
|
||||
func effectiveNeo4jScheme(params connectParams) string {
|
||||
if value := strings.TrimSpace(params.ConnectionString); value != "" {
|
||||
parsed, err := url.Parse(strings.TrimPrefix(value, "jdbc:"))
|
||||
if err == nil && isNeo4jScheme(parsed.Scheme) && parsed.Host != "" {
|
||||
return strings.ToLower(parsed.Scheme)
|
||||
}
|
||||
}
|
||||
query := parseURLParams(params.URLParams)
|
||||
if configured := strings.ToLower(strings.TrimSpace(query.Get("scheme"))); isNeo4jScheme(configured) {
|
||||
scheme = configured
|
||||
} else if params.SSL || queryBool(query, "ssl") || queryBool(query, "encrypted") || params.CACertPath != "" || params.ClientCertPath != "" {
|
||||
scheme = "neo4j+s"
|
||||
return configured
|
||||
}
|
||||
return (&url.URL{Scheme: scheme, Host: net.JoinHostPort(host, strconv.Itoa(port))}).String(), nil
|
||||
if neo4jEncryptionRequested(params, query) {
|
||||
return "neo4j+s"
|
||||
}
|
||||
return "neo4j"
|
||||
}
|
||||
|
||||
func neo4jEncryptionRequested(params connectParams, query url.Values) bool {
|
||||
return params.SSL || queryBool(query, "ssl") || queryBool(query, "encrypted") ||
|
||||
params.CACertPath != "" || params.ClientCertPath != ""
|
||||
}
|
||||
|
||||
// directConnectionScheme maps a routing scheme to its direct counterpart. The
|
||||
// second result is false for schemes that already connect directly.
|
||||
func directConnectionScheme(scheme string) (string, bool) {
|
||||
switch strings.ToLower(strings.TrimSpace(scheme)) {
|
||||
case "neo4j":
|
||||
return "bolt", true
|
||||
case "neo4j+s":
|
||||
return "bolt+s", true
|
||||
case "neo4j+ssc":
|
||||
return "bolt+ssc", true
|
||||
default:
|
||||
return "", false
|
||||
}
|
||||
}
|
||||
|
||||
// directConnectionParams rewrites routing-style settings (`neo4j://`) into the
|
||||
// equivalent direct settings (`bolt://`), keeping TLS semantics. The second
|
||||
// result is false when the connection is already direct or cannot be rewritten.
|
||||
func directConnectionParams(params connectParams) (connectParams, bool) {
|
||||
direct, ok := directConnectionScheme(effectiveNeo4jScheme(params))
|
||||
if !ok {
|
||||
return params, false
|
||||
}
|
||||
if value := strings.TrimSpace(params.ConnectionString); value != "" {
|
||||
parsed, err := url.Parse(strings.TrimPrefix(value, "jdbc:"))
|
||||
if err != nil || !isNeo4jScheme(parsed.Scheme) || parsed.Host == "" {
|
||||
return params, false
|
||||
}
|
||||
parsed.Scheme = direct
|
||||
parsed.User = nil
|
||||
parsed.Path = ""
|
||||
parsed.RawPath = ""
|
||||
parsed.RawQuery = filterDriverURIQuery(parsed.Query()).Encode()
|
||||
params.ConnectionString = parsed.String()
|
||||
return params, true
|
||||
}
|
||||
if strings.TrimSpace(params.Host) == "" {
|
||||
return params, false
|
||||
}
|
||||
params.URLParams = withURLParam(params.URLParams, "scheme", direct)
|
||||
return params, true
|
||||
}
|
||||
|
||||
// routingUnsupportedMessage is the wording the Neo4j driver uses when a Bolt 3
|
||||
// server reports that cluster routing is unavailable.
|
||||
const routingUnsupportedMessage = "does not support: routing"
|
||||
|
||||
// isRoutingUnsupported reports whether err is the driver's "routing requires a
|
||||
// cluster" failure. The driver flattens its typed feature error into a usage
|
||||
// error, so the message is inspected as well as the type.
|
||||
func isRoutingUnsupported(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
var featureErr *neo4jdb.FeatureNotSupportedError
|
||||
if errors.As(err, &featureErr) && strings.EqualFold(featureErr.Feature, "routing") {
|
||||
return true
|
||||
}
|
||||
return strings.Contains(strings.ToLower(err.Error()), routingUnsupportedMessage)
|
||||
}
|
||||
|
||||
func isNeo4jScheme(value string) bool {
|
||||
@@ -180,15 +294,33 @@ func filterDriverURIQuery(values url.Values) url.Values {
|
||||
return result
|
||||
}
|
||||
|
||||
// withURLParam replaces one `key=value` entry in a URL parameter string while
|
||||
// keeping the remaining entries untouched. Values are copied verbatim (unlike
|
||||
// url.Values.Encode) so schemes such as `bolt+s` are not percent-encoded.
|
||||
func withURLParam(urlParams, key, value string) string {
|
||||
parts := make([]string, 0, 4)
|
||||
for _, pair := range strings.FieldsFunc(urlParams, func(char rune) bool { return char == '&' || char == ';' }) {
|
||||
pair = strings.TrimSpace(pair)
|
||||
if pair == "" {
|
||||
continue
|
||||
}
|
||||
existingKey, _, _ := strings.Cut(pair, "=")
|
||||
if strings.EqualFold(strings.TrimSpace(existingKey), key) {
|
||||
continue
|
||||
}
|
||||
parts = append(parts, pair)
|
||||
}
|
||||
parts = append(parts, key+"="+value)
|
||||
return strings.Join(parts, "&")
|
||||
}
|
||||
|
||||
func testConnection(params connectParams) error {
|
||||
driver, err := openDriver(params)
|
||||
runtime, err := newConnectionRuntime(params)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer driver.Close(context.Background())
|
||||
ctx, cancel := context.WithTimeout(context.Background(), defaultConnectTimeout)
|
||||
defer cancel()
|
||||
return driver.VerifyConnectivity(ctx)
|
||||
_ = runtime.close()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *server) validateConnection() error {
|
||||
@@ -201,6 +333,11 @@ func (s *server) validateConnection() error {
|
||||
}
|
||||
|
||||
func (s *server) databaseName(override string) string {
|
||||
if s.runtime != nil && s.runtime.legacySingleDatabase {
|
||||
// Servers that negotiated Bolt 3 only expose the default database and
|
||||
// reject any explicit database name, even the configured one.
|
||||
return ""
|
||||
}
|
||||
if value := strings.TrimSpace(override); value != "" {
|
||||
return value
|
||||
}
|
||||
|
||||
@@ -45,6 +45,38 @@ func TestLiveNeo4jAgent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestLiveNeo4jAgentSingleInstanceRoutingFallback covers single-instance
|
||||
// servers that cannot serve `neo4j://` routing. The connection intentionally
|
||||
// keeps the default routing scheme so the direct-connection fallback runs.
|
||||
func TestLiveNeo4jAgentSingleInstanceRoutingFallback(t *testing.T) {
|
||||
if os.Getenv("DBX_NEO4J_LIVE") != "1" {
|
||||
t.Skip("set DBX_NEO4J_LIVE=1 to run against a real Neo4j server")
|
||||
}
|
||||
params := connectParams{
|
||||
Host: envOr("DBX_NEO4J_HOST", "127.0.0.1"),
|
||||
Port: envIntOr("DBX_NEO4J_PORT", defaultNeo4jPort),
|
||||
Database: envOr("DBX_NEO4J_DATABASE", defaultDatabase),
|
||||
Username: envOr("DBX_NEO4J_USER", "neo4j"),
|
||||
Password: os.Getenv("DBX_NEO4J_PASSWORD"),
|
||||
}
|
||||
runtime, err := newConnectionRuntime(params)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer runtime.close()
|
||||
if runtime.legacySingleDatabase && runtime.params.URLParams != "scheme=bolt" {
|
||||
t.Fatalf("legacy server must fall back to a direct connection, got %q", runtime.params.URLParams)
|
||||
}
|
||||
server := newServer(runtime, params)
|
||||
result, err := server.executeQuery(queryOptions{SQL: "RETURN 1 AS value", MaxRows: 10})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(result.Rows) != 1 || result.Rows[0][0] != "1" {
|
||||
t.Fatalf("unexpected query result: %#v", result)
|
||||
}
|
||||
}
|
||||
|
||||
func rawParams(values map[string]any) map[string]json.RawMessage {
|
||||
result := make(map[string]json.RawMessage, len(values))
|
||||
for key, value := range values {
|
||||
|
||||
@@ -92,11 +92,14 @@ type querySession struct {
|
||||
}
|
||||
|
||||
type connectionRuntime struct {
|
||||
driver neo4j.Driver
|
||||
params connectParams
|
||||
references int
|
||||
closed bool
|
||||
mu sync.Mutex
|
||||
driver neo4j.Driver
|
||||
params connectParams
|
||||
// legacySingleDatabase is set for servers that negotiated Bolt 3 or older:
|
||||
// they have a single default database and reject session database selection.
|
||||
legacySingleDatabase bool
|
||||
references int
|
||||
closed bool
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
type server struct {
|
||||
|
||||
@@ -3,6 +3,8 @@ package main
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/url"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
@@ -78,6 +80,132 @@ func TestConfiguredDatabaseUsesConnectionString(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestEffectiveNeo4jSchemeMatchesBuiltURI(t *testing.T) {
|
||||
for _, params := range []connectParams{
|
||||
{Host: "127.0.0.1", Port: 7687},
|
||||
{Host: "db.example.com", Port: 7687, SSL: true},
|
||||
{Host: "db.example.com", Port: 7687, URLParams: "encrypted=true"},
|
||||
{Host: "db", Port: 7687, URLParams: "scheme=bolt"},
|
||||
{Host: "db", Port: 7687, URLParams: "scheme=neo4j+ssc"},
|
||||
{ConnectionString: "jdbc:neo4j://user:secret@db:7687?database=movies"},
|
||||
{ConnectionString: "bolt://db:7687"},
|
||||
} {
|
||||
uri, err := buildNeo4jURI(params)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
parsed, err := url.Parse(uri)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := effectiveNeo4jScheme(params); got != parsed.Scheme {
|
||||
t.Fatalf("effectiveNeo4jScheme(%#v) = %q, want %q", params, got, parsed.Scheme)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestDirectConnectionParamsRewritesRoutingSchemes(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
params connectParams
|
||||
want connectParams
|
||||
wantChanged bool
|
||||
}{
|
||||
{
|
||||
name: "default routing scheme becomes direct",
|
||||
params: connectParams{Host: "db", Port: 7687},
|
||||
want: connectParams{Host: "db", Port: 7687, URLParams: "scheme=bolt"},
|
||||
wantChanged: true,
|
||||
},
|
||||
{
|
||||
name: "tls keeps encryption",
|
||||
params: connectParams{Host: "db", Port: 7687, SSL: true, URLParams: "connection_timeout=5s"},
|
||||
want: connectParams{Host: "db", Port: 7687, SSL: true, URLParams: "connection_timeout=5s&scheme=bolt+s"},
|
||||
wantChanged: true,
|
||||
},
|
||||
{
|
||||
name: "explicit routing scheme is rewritten in place",
|
||||
params: connectParams{Host: "db", Port: 7687, URLParams: "scheme=neo4j&database=movies"},
|
||||
want: connectParams{Host: "db", Port: 7687, URLParams: "database=movies&scheme=bolt"},
|
||||
wantChanged: true,
|
||||
},
|
||||
{
|
||||
name: "direct connection is kept",
|
||||
params: connectParams{Host: "db", Port: 7687, URLParams: "scheme=bolt"},
|
||||
want: connectParams{Host: "db", Port: 7687, URLParams: "scheme=bolt"},
|
||||
wantChanged: false,
|
||||
},
|
||||
{
|
||||
name: "routing connection string becomes direct",
|
||||
params: connectParams{ConnectionString: "jdbc:neo4j://user:secret@db:7687?database=movies"},
|
||||
want: connectParams{ConnectionString: "bolt://db:7687"},
|
||||
wantChanged: true,
|
||||
},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
got, changed := directConnectionParams(testCase.params)
|
||||
if changed != testCase.wantChanged {
|
||||
t.Fatalf("directConnectionParams() changed = %t, want %t", changed, testCase.wantChanged)
|
||||
}
|
||||
if !changed {
|
||||
return
|
||||
}
|
||||
uri, err := buildNeo4jURI(got)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wantURI, err := buildNeo4jURI(testCase.want)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if uri != wantURI {
|
||||
t.Fatalf("directConnectionParams() built %q, want %q", uri, wantURI)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsRoutingUnsupported(t *testing.T) {
|
||||
for _, testCase := range []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{
|
||||
name: "driver usage error wording",
|
||||
err: errors.New("feature not supported: Server 127.0.0.1:17690 does not support: routing (requires cluster setup)"),
|
||||
want: true,
|
||||
},
|
||||
{name: "typed feature error", err: &neo4jdb.FeatureNotSupportedError{Server: "db:7687", Feature: "routing", Reason: "requires cluster setup"}, want: true},
|
||||
{name: "typed route-to-database error", err: &neo4jdb.FeatureNotSupportedError{Server: "db:7687", Feature: "route to database", Reason: "requires at least server v4"}, want: false},
|
||||
{name: "unrelated error", err: errors.New("connection refused"), want: false},
|
||||
{name: "nil error", err: nil, want: false},
|
||||
} {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
if got := isRoutingUnsupported(testCase.err); got != testCase.want {
|
||||
t.Fatalf("isRoutingUnsupported(%v) = %t, want %t", testCase.err, got, testCase.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestWithURLParamKeepsOtherEntries(t *testing.T) {
|
||||
cases := []struct {
|
||||
urlParams string
|
||||
want string
|
||||
}{
|
||||
{urlParams: "", want: "scheme=bolt+s"},
|
||||
{urlParams: "scheme=neo4j", want: "scheme=bolt+s"},
|
||||
{urlParams: "max_connection_pool_size=10;scheme=neo4j", want: "max_connection_pool_size=10&scheme=bolt+s"},
|
||||
{urlParams: "a=1&b=2", want: "a=1&b=2&scheme=bolt+s"},
|
||||
}
|
||||
for _, testCase := range cases {
|
||||
if got := withURLParam(testCase.urlParams, "scheme", "bolt+s"); got != testCase.want {
|
||||
t.Fatalf("withURLParam(%q) = %q, want %q", testCase.urlParams, got, testCase.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMetadataWindowAndFiltering(t *testing.T) {
|
||||
values := []string{"alpha", "beta", "gamma"}
|
||||
if got := applyMetadataWindow(values, 1, 1); !reflect.DeepEqual(got, []string{"beta"}) {
|
||||
|
||||
Reference in New Issue
Block a user