Devshard Gateway Postgres Backend

decentralized-api/payloadstorage/factory.go

@@ -28,14 +28,18 @@ func NewPayloadStorage(ctx context.Context, fileBasePath string) PayloadStorage
 	if err != nil || retryInterval <= 0 {
 		retryInterval = 240 * time.Second
 	}
+	connectTimeout, err := time.ParseDuration(os.Getenv("PG_CONNECT_TIMEOUT"))
+	if err != nil || connectTimeout <= 0 {
+		connectTimeout = defaultPGConnectTimeout
+	}
 
 	pgStorage, err := NewPostgresStorage(ctx)
 	if err != nil {
 		logging.Warn("PostgreSQL connection failed, will retry lazily on Store", types.PayloadStorage,
 			"host", pgHost, "error", err)
-		return NewHybridStorage(nil, fileStorage, retryInterval)
+		return NewHybridStorage(nil, fileStorage, retryInterval, connectTimeout)
 	}
 
 	logging.Info("Using PostgreSQL with file fallback", types.PayloadStorage, "host", pgHost)
-	return NewHybridStorage(pgStorage, fileStorage, retryInterval)
+	return NewHybridStorage(pgStorage, fileStorage, retryInterval, connectTimeout)
 }

decentralized-api/payloadstorage/factory_test.go

@@ -76,3 +76,51 @@ func TestNewPayloadStorage_RetryInterval(t *testing.T) {
 	}
 }
 
+func TestNewPayloadStorage_ConnectTimeout(t *testing.T) {
+	os.Setenv("PGHOST", "localhost")
+	defer os.Unsetenv("PGHOST")
+
+	tests := []struct {
+		name     string
+		envValue string
+		expected time.Duration
+	}{
+		{
+			name:     "Default (unset)",
+			envValue: "",
+			expected: defaultPGConnectTimeout,
+		},
+		{
+			name:     "Custom valid duration",
+			envValue: "500ms",
+			expected: 500 * time.Millisecond,
+		},
+		{
+			name:     "Invalid duration (fallback to default)",
+			envValue: "invalid",
+			expected: defaultPGConnectTimeout,
+		},
+		{
+			name:     "Zero duration (fallback to default)",
+			envValue: "0s",
+			expected: defaultPGConnectTimeout,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.envValue != "" {
+				os.Setenv("PG_CONNECT_TIMEOUT", tt.envValue)
+				defer os.Unsetenv("PG_CONNECT_TIMEOUT")
+			} else {
+				os.Unsetenv("PG_CONNECT_TIMEOUT")
+			}
+
+			s := NewPayloadStorage(context.Background(), t.TempDir())
+			hs, ok := s.(*HybridStorage)
+			require.True(t, ok, "Expected *HybridStorage")
+			assert.Equal(t, tt.expected, hs.connectTimeout)
+		})
+	}
+}
+

decentralized-api/payloadstorage/hybrid_storage.go

@@ -11,24 +11,34 @@ import (
 	"github.com/productscience/inference/x/inference/types"
 )
 
-const (
-	pgConnectTimeout = 2 * time.Second
-)
+const defaultPGConnectTimeout = 2 * time.Second
 
 // HybridStorage uses PostgreSQL as primary storage with file-based fallback.
 // Store: tries PG first (with lazy reconnection), falls back to file on error.
 // Retrieve: tries PG first (no reconnection delay), on error OR not found also checks file.
 // PruneEpoch: prunes both (best effort, no reconnection delay).
 type HybridStorage struct {
-	pg            *PostgresStorage
-	file          *FileStorage
-	mu            sync.Mutex
-	lastRetry     time.Time
-	retryInterval time.Duration
+	pg             *PostgresStorage
+	file           *FileStorage
+	mu             sync.Mutex
+	lastRetry      time.Time
+	retryInterval  time.Duration
+	connectTimeout time.Duration
 }
 
-func NewHybridStorage(pg *PostgresStorage, file *FileStorage, retryInterval time.Duration) *HybridStorage {
-	return &HybridStorage{pg: pg, file: file, retryInterval: retryInterval}
+func NewHybridStorage(pg *PostgresStorage, file *FileStorage, retryInterval, connectTimeout time.Duration) *HybridStorage {
+	if retryInterval <= 0 {
+		retryInterval = 240 * time.Second
+	}
+	if connectTimeout <= 0 {
+		connectTimeout = defaultPGConnectTimeout
+	}
+	return &HybridStorage{
+		pg:             pg,
+		file:           file,
+		retryInterval:  retryInterval,
+		connectTimeout: connectTimeout,
+	}
 }
 
 // shouldAttemptConnect checks if reconnection should be attempted.
@@ -66,7 +76,7 @@ func (h *HybridStorage) getOrConnectPg(ctx context.Context) *PostgresStorage {
 		return pg
 	}
 
-	connectCtx, cancel := context.WithTimeout(ctx, pgConnectTimeout)
+	connectCtx, cancel := context.WithTimeout(ctx, h.connectTimeout)
 	defer cancel()
 
 	newPg, err := NewPostgresStorage(connectCtx)

decentralized-api/payloadstorage/postgres_storage_test.go

@@ -228,7 +228,7 @@ func TestHybridStorage_FallbackOnPGError(t *testing.T) {
 	require.NoError(t, err)
 	defer pgStorage.Close()
 
-	hybrid := NewHybridStorage(pgStorage, fileStorage, 240*time.Second)
+	hybrid := NewHybridStorage(pgStorage, fileStorage, 240*time.Second, defaultPGConnectTimeout)
 
 	// Data not in PG, but is in file - should find it
 	prompt, response, err := hybrid.Retrieve(ctx, "inf-001", 100)
@@ -250,7 +250,7 @@ func TestHybridStorage_PGPrimary(t *testing.T) {
 	defer pgStorage.Close()
 
 	fileStorage := NewFileStorage(tempDir)
-	hybrid := NewHybridStorage(pgStorage, fileStorage, 240*time.Second)
+	hybrid := NewHybridStorage(pgStorage, fileStorage, 240*time.Second, defaultPGConnectTimeout)
 
 	// Store via hybrid (should go to PG)
 	storedPrompt := []byte(`{"pg": "data"}`)

devshard/cmd/devshardctl/capacity_state_test.go

@@ -367,7 +367,7 @@ func TestGatewayLimiterUnlimitedBaselinePreservedUnderScale(t *testing.T) {
 
 func TestDevshardRuntimeLoadIsActivePerWeight(t *testing.T) {
 	rt := &devshardRuntime{}
-	rt.activeRequests.Store(2)
+	rt.activeUserRequests.Store(2)
 	rt.reservedTokens.Store(3) // reserved tokens no longer factor in.
 
 	require.InDelta(t, 0.5, rt.load(4.0), 1e-9)

devshard/cmd/devshardctl/chain_tx_rest.go

@@ -3,7 +3,9 @@ package main
 import (
 	"bytes"
 	"context"
+	"crypto/sha256"
 	"encoding/base64"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -113,7 +115,49 @@ func NewRESTChainTxClient(cfg RESTChainTxConfig) (*RESTChainTxClient, error) {
 	}, nil
 }
 
-func (c *RESTChainTxClient) CreateDevshardEscrow(ctx context.Context, signer *signing.Secp256k1Signer, amount uint64, modelID string) (*CreateDevshardEscrowResult, error) {
+// txHashFromBytes returns the cosmos tx hash (uppercase-hex SHA-256 of the tx
+// bytes), computable before broadcast and equal to the hash the node returns.
+func txHashFromBytes(txBytes []byte) string {
+	sum := sha256.Sum256(txBytes)
+	return strings.ToUpper(hex.EncodeToString(sum[:]))
+}
+
+// errTxNotFound marks a tx that is not on chain (404) — distinct from one that
+// committed but failed. The tx may still land until its unordered TTL elapses.
+var errTxNotFound = errors.New("tx not found on chain")
+
+// GetTxEscrowID resolves a create tx hash to its escrow_id in one lookup;
+// found=false when the tx is absent or failed (no escrow), error when unreachable.
+func (c *RESTChainTxClient) GetTxEscrowID(ctx context.Context, txHash string) (uint64, bool, error) {
+	var lastErr error
+	hasNotFoundError := false
+	for _, baseURL := range c.txQueryBaseURLs() {
+		var payload txResponseEnvelope
+		err := c.getJSONFromBaseURL(ctx, baseURL, "/cosmos/tx/v1beta1/txs/"+url.PathEscape(txHash), &payload)
+		if err != nil {
+			if isNotFoundError(err) {
+				hasNotFoundError = true // this endpoint lacks it; try the fallback before deciding
+				continue
+			}
+			lastErr = fmt.Errorf("%s: %w", baseURL, err)
+			continue
+		}
+		if payload.TxResponse.Code != 0 {
+			return 0, false, nil // tx committed but failed → no escrow created
+		}
+		if escrowID, ok := payload.TxResponse.createdEscrowID(); ok {
+			return escrowID, true, nil
+		}
+		lastErr = fmt.Errorf("tx %s committed via %s but escrow_id event was not found", txHash, baseURL)
+	}
+	// Only conclude "not on chain" when every reachable endpoint agreed on 404.
+	if lastErr == nil && hasNotFoundError {
+		return 0, false, errTxNotFound
+	}
+	return 0, false, lastErr
+}
+
+func (c *RESTChainTxClient) CreateDevshardEscrow(ctx context.Context, signer *signing.Secp256k1Signer, amount uint64, modelID string, onPrepared func(txHash string) error) (*CreateDevshardEscrowResult, error) {
 	if c == nil {
 		return nil, fmt.Errorf("chain tx client is nil")
 	}
@@ -145,10 +189,20 @@ func (c *RESTChainTxClient) CreateDevshardEscrow(ctx context.Context, signer *si
 	if err != nil {
 		return nil, err
 	}
-	txHash, err := c.broadcastTx(ctx, txBytes)
+	// Record the intent (precomputed hash) before the irreversible broadcast; abort if it fails.
+	txHash := txHashFromBytes(txBytes)
+	if onPrepared != nil {
+		if err := onPrepared(txHash); err != nil {
+			return nil, fmt.Errorf("record escrow create intent before broadcast: %w", err)
+		}
+	}
+	broadcastHash, err := c.broadcastTx(ctx, txBytes)
 	if err != nil {
 		return nil, err
 	}
+	if !strings.EqualFold(broadcastHash, txHash) {
+		return nil, fmt.Errorf("tx hash mismatch: precomputed %s, node returned %s", txHash, broadcastHash)
+	}
 	escrowID, err := c.waitForCreatedEscrowID(ctx, txHash)
 	if err != nil {
 		return nil, err
@@ -332,6 +386,16 @@ func (c *RESTChainTxClient) getJSONFromBaseURL(ctx context.Context, baseURL, pat
 	return json.NewDecoder(resp.Body).Decode(out)
 }
 
+// isNotFoundError reports whether a chain GET failed with 404 / not-found rather
+// than a transient error.
+func isNotFoundError(err error) bool {
+	if err == nil {
+		return false
+	}
+	s := strings.ToLower(err.Error())
+	return strings.Contains(s, "status 404") || strings.Contains(s, "not found")
+}
+
 func (c *RESTChainTxClient) postJSON(ctx context.Context, path string, in any, out any) error {
 	data, err := json.Marshal(in)
 	if err != nil {