Better backend and cache performance

ARCHITECTURE.md

@@ -278,6 +278,14 @@ Concrete example for config:
 6. It queues a `lastSeenUpdate` in the batch writer.
 7. It returns `env.Configuration` as the osquery config payload.
 
+### Steady-state database load
+
+- Node check-ins are coalesced per TLS process and persisted in parameterized bulk updates of at most 100 nodes. Each node retains its observation timestamp, normalized to database precision; older observations cannot replace newer timestamps or IPs. Equal stored timestamps keep the first persisted IP. The existing writer batch size, timeout, and buffer settings still control collection. The queue remains process-local and best-effort on shutdown or database errors; it is not a durable heartbeat store.
+- Empty distributed-query results are cached in Redis for two minutes by default, configurable through `osquery.queryDispatchTTL`, `--query-dispatch-ttl`, or `QUERY_DISPATCH_TTL`. Non-positive values select the default. Query creation invalidates targeted nodes; Redis `WATCH` prevents an in-flight empty SQL result from refilling an invalidated entry. SQL errors are not cached, and TLS returns HTTP 503 so agents can retry.
+- When acceleration is disabled, query reads skip session checks entirely. Otherwise, shared console/file-explorer hints cache absence for two minutes and presence for at most five seconds, bounded by the existing 30-second session freshness window. Session mutations invalidate hints after committing; token-checked refills reject obsolete lookups. These hints control polling only, never authorization.
+- Redis failures fall back to SQL. Failed invalidations can delay query discovery by the configured dispatch TTL or session acceleration by two minutes. Direct-database CLI mode has no Redis connection and also relies on dispatch TTL expiry; use API-mode query/carve submission for immediate invalidation. Upgrade all API/TLS replicas before relying on race-safe invalidation; old replicas do not implement the new refill protocol. Longer TTLs trade fewer SQL reads for a larger failure-time staleness window.
+- Real-engine regression tests are opt-in: `OSCTRL_TEST_POSTGRES_DSN` and `OSCTRL_TEST_MYSQL_DSN` exercise bulk updates using temporary prefixed tables. `OSCTRL_TEST_REDIS_ADDR` exercises query/session cache races against a disposable Redis instance; the optional `OSCTRL_TEST_REDIS_CONTAINER` test restarts that disposable container and requires a fixed host port.
+
 ### operator/API flow
 
 ```text

cmd/api/handlers/queries.go

@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"context"
 	"encoding/csv"
 	"encoding/json"
 	"fmt"
@@ -255,6 +256,8 @@ func (h *HandlersApi) QueriesRunHandler(w http.ResponseWriter, r *http.Request)
 		apiErrorResponse(w, "error creating query", http.StatusInternalServerError, err)
 		return
 	}
+	// The commit is durable even if the HTTP request was canceled.
+	h.Queries.Cache.InvalidateMany(context.Background(), targetNodesID)
 	// Return query name as serialized response
 	log.Debug().Msgf("Created query %s with id %d", newQuery.Name, newQuery.ID)
 	h.AuditLog.NewQuery(ctx[ctxUser], newQuery.Query, strings.Split(r.RemoteAddr, ":")[0], env.ID)

cmd/api/handlers/query_dispatch_test.go

@@ -0,0 +1,89 @@
+package handlers
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+
+	redis "github.com/go-redis/redis/v8"
+	"github.com/jmpsec/osctrl/pkg/auditlog"
+	"github.com/jmpsec/osctrl/pkg/config"
+	"github.com/jmpsec/osctrl/pkg/queries"
+	"github.com/jmpsec/osctrl/pkg/types"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+)
+
+func TestCreateQueryInvalidatesDispatchAfterCommit(t *testing.T) {
+	addr := os.Getenv("OSCTRL_TEST_REDIS_ADDR")
+	if addr == "" {
+		t.Skip("set OSCTRL_TEST_REDIS_ADDR to a disposable Redis instance")
+	}
+	for _, name := range []string{"commit", "rollback", "canceled-request"} {
+		t.Run(name, func(t *testing.T) {
+			rollback := name == "rollback"
+			db, h, env, node := setupConsoleHandlers(t)
+			h.AuditLog = &auditlog.AuditLogManager{}
+			h.DebugHTTPConfig = &config.YAMLConfigurationDebug{}
+			client := redis.NewClient(&redis.Options{Addr: addr})
+			t.Cleanup(func() { _ = client.Close() })
+			h.Queries.Cache = queries.NewQueryDispatchCache(client, 0)
+			h.Queries.Cache.Invalidate(context.Background(), node.ID)
+			t.Cleanup(func() { h.Queries.Cache.Invalidate(context.Background(), node.ID) })
+			result, _, err := h.Queries.NodeQueries(node)
+			require.NoError(t, err)
+			require.Empty(t, result)
+			cached, err := h.Queries.Cache.HasNoPendingQueries(context.Background(), node.ID)
+			require.NoError(t, err)
+			require.True(t, cached)
+
+			body, err := json.Marshal(types.ApiDistributedQueryRequest{Query: "SELECT 1", UUIDs: []string{node.UUID}})
+			require.NoError(t, err)
+			req := consoleRequest(http.MethodPost, "/queries", body, "alice")
+			requestCtx, cancel := context.WithCancel(req.Context())
+			defer cancel()
+			req = req.WithContext(requestCtx)
+			req.SetPathValue("env", env.Name)
+			checkedBeforeCommit := false
+			require.NoError(t, db.Callback().Create().Before("gorm:create").Register("test:before_target", func(tx *gorm.DB) {
+				if tx.Statement.Table != "distributed_query_targets" {
+					return
+				}
+				checkedBeforeCommit = true
+				cached, err := h.Queries.Cache.HasNoPendingQueries(context.Background(), node.ID)
+				require.NoError(t, err)
+				require.True(t, cached, "do not invalidate while query creation is uncommitted")
+				if rollback {
+					tx.AddError(errors.New("target write failed"))
+				}
+				if name == "canceled-request" {
+					cancel()
+				}
+			}))
+			rr := httptest.NewRecorder()
+			h.QueriesRunHandler(rr, req)
+			require.True(t, checkedBeforeCommit, rr.Body.String())
+			cached, err = h.Queries.Cache.HasNoPendingQueries(context.Background(), node.ID)
+			require.NoError(t, err)
+			if rollback {
+				require.Equal(t, http.StatusInternalServerError, rr.Code, rr.Body.String())
+				require.True(t, cached, "rollback must preserve the idle hint")
+				var count int64
+				require.NoError(t, db.Model(&queries.NodeQuery{}).Count(&count).Error)
+				require.Zero(t, count)
+				return
+			}
+			require.Equal(t, http.StatusOK, rr.Code, rr.Body.String())
+			require.False(t, cached, "commit must invalidate the idle hint")
+			var response types.ApiQueriesResponse
+			require.NoError(t, json.Unmarshal(rr.Body.Bytes(), &response))
+			result, _, err = h.Queries.NodeQueries(node)
+			require.NoError(t, err)
+			require.Equal(t, queries.QueryReadQueries{response.Name: "SELECT 1"}, result)
+		})
+	}
+}

cmd/api/main.go

@@ -450,8 +450,11 @@ func osctrlAPIService() {
 	queriesmgr.Cache = queries.NewQueryDispatchCache(redis.Client, 0)
 	log.Info().Msg("Initialize console")
 	consolemgr = console.NewManager(db.Conn, queriesmgr)
+	sessionHints := cache.NewSessionHints(redis.Client)
+	consolemgr.SessionHints = sessionHints
 	log.Info().Msg("Initialize file explorer")
 	fileexplorermgr = fileexplorer.NewManager(db.Conn, queriesmgr)
+	fileexplorermgr.SessionHints = sessionHints
 	// Construct the log reader. When the TLS logger ships logs to S3 the
 	// osquery_*_data tables are empty, so the API/console/file-explorer
 	// read logs back from S3 instead. For every other logger the data

cmd/tls/config_test.go

@@ -1,14 +1,50 @@
 package main
 
 import (
+	"context"
 	"os"
 	"path/filepath"
 	"testing"
 	"time"
 
 	"github.com/jmpsec/osctrl/pkg/config"
+	"github.com/urfave/cli/v3"
 )
 
+func TestQueryDispatchTTLConfiguration(t *testing.T) {
+	cfg, err := loadYAMLConfiguration(writeTempTLSConfig(t, "service:\n  auth: none\nosquery:\n  queryDispatchTTL: 45s\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	params := loadedYAMLToServiceParams(cfg, "tls.yml")
+	if got := params.Osquery.QueryDispatchTTL; got != 45*time.Second {
+		t.Fatalf("YAML queryDispatchTTL = %s, want 45s", got)
+	}
+
+	var dispatchFlag *cli.DurationFlag
+	for _, flag := range flags {
+		if f, ok := flag.(*cli.DurationFlag); ok && f.Name == "query-dispatch-ttl" {
+			copy := *f
+			dispatchFlag = &copy
+		}
+	}
+	if dispatchFlag == nil {
+		t.Fatal("missing query-dispatch-ttl flag")
+	}
+	if dispatchFlag.Destination != &flagParams.Osquery.QueryDispatchTTL {
+		t.Fatal("query-dispatch-ttl flag has wrong destination")
+	}
+	dispatchFlag.Destination = &params.Osquery.QueryDispatchTTL
+	t.Setenv("QUERY_DISPATCH_TTL", "90s")
+	command := &cli.Command{Flags: []cli.Flag{dispatchFlag}, Action: func(context.Context, *cli.Command) error { return nil }}
+	if err := command.Run(context.Background(), []string{"tls"}); err != nil {
+		t.Fatal(err)
+	}
+	if got := params.Osquery.QueryDispatchTTL; got != 90*time.Second {
+		t.Fatalf("environment queryDispatchTTL = %s, want 90s", got)
+	}
+}
+
 func writeTempTLSConfig(t *testing.T, body string) string {
 	t.Helper()
 	path := filepath.Join(t.TempDir(), "tls.yml")

cmd/tls/handlers/console_acceleration_test.go

@@ -25,7 +25,7 @@ func TestShouldAccelerateQueryReadForActiveConsoleSession(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, db.AutoMigrate(&console.Session{}))
 	queryManager := queries.CreateQueries(db)
-	handler := &HandlersTLS{Queries: queryManager, OsqueryValues: &config.YAMLConfigurationOsquery{Console: true}}
+	handler := &HandlersTLS{Queries: queryManager, OsqueryValues: &config.YAMLConfigurationOsquery{Accelerated: true, Console: true}}
 	node := nodes.OsqueryNode{ID: 7, UUID: "NODE-UUID", EnvironmentID: 1}
 	otherNode := nodes.OsqueryNode{ID: 8, UUID: "OTHER-NODE-UUID", EnvironmentID: 1}
 
@@ -96,7 +96,7 @@ func TestShouldNotAccelerateQueryReadForConsoleWhenConsoleDisabled(t *testing.T)
 	require.NoError(t, err)
 	require.NoError(t, db.AutoMigrate(&console.Session{}))
 	queryManager := queries.CreateQueries(db)
-	handler := &HandlersTLS{Queries: queryManager, OsqueryValues: &config.YAMLConfigurationOsquery{}}
+	handler := &HandlersTLS{Queries: queryManager, OsqueryValues: &config.YAMLConfigurationOsquery{Accelerated: true}}
 	node := nodes.OsqueryNode{ID: 7, UUID: "NODE-UUID", EnvironmentID: 1}
 
 	require.NoError(t, db.Create(&console.Session{
@@ -130,7 +130,7 @@ func TestShouldAccelerateQueryReadForActiveFileExplorerSession(t *testing.T) {
 	}).Error)
 
 	require.False(t, handler.shouldAccelerateQueryRead(node, false))
-	handler.OsqueryValues = &config.YAMLConfigurationOsquery{FileExplorer: true}
+	handler.OsqueryValues = &config.YAMLConfigurationOsquery{Accelerated: true, FileExplorer: true}
 	require.True(t, handler.shouldAccelerateQueryRead(node, false))
 }
 
@@ -204,3 +204,18 @@ func queryReadResponse(t *testing.T, handler *HandlersTLS, envUUID, nodeKey stri
 	require.NoError(t, json.Unmarshal(rr.Body.Bytes(), &resp))
 	return resp
 }
+
+func TestDisabledAccelerationSkipsSessionSQL(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	require.NoError(t, err)
+	require.NoError(t, db.AutoMigrate(&console.Session{}, &fileexplorer.Session{}))
+	reads := 0
+	require.NoError(t, db.Callback().Query().Before("gorm:query").Register("count_session_reads", func(*gorm.DB) { reads++ }))
+	h := &HandlersTLS{Queries: &queries.Queries{DB: db}, OsqueryValues: &config.YAMLConfigurationOsquery{Console: true, FileExplorer: true}}
+	node := nodes.OsqueryNode{ID: 7, UUID: "NODE-UUID", EnvironmentID: 1}
+	require.False(t, h.shouldAccelerateQueryRead(node, false))
+	require.False(t, h.shouldAccelerateQueryRead(node, true))
+	require.Zero(t, reads, "disabled acceleration must not query session tables")
+	h.OsqueryValues = nil
+	require.False(t, h.shouldAccelerateQueryRead(node, true))
+}

cmd/tls/handlers/handlers.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/jmpsec/osctrl/pkg/auditlog"
 	"github.com/jmpsec/osctrl/pkg/backend"
+	"github.com/jmpsec/osctrl/pkg/cache"
 	"github.com/jmpsec/osctrl/pkg/carves"
 	"github.com/jmpsec/osctrl/pkg/config"
 	"github.com/jmpsec/osctrl/pkg/console"
@@ -38,7 +39,7 @@ var validAction = map[string]bool{
 	settings.ScriptRemove: true,
 }
 
-const consoleSessionFreshness = 30 * time.Second
+const consoleSessionFreshness = cache.SessionFreshness
 
 // Valid values for enroll packages
 var validEnrollPackage = map[string]bool{
@@ -65,6 +66,7 @@ type HandlersTLS struct {
 	Carves          *carves.Carves
 	Settings        *settings.Settings
 	SettingsCache   *settings.RedisSettingsCache
+	SessionHints    *cache.SessionHints
 	Logs            *logging.LoggerTLS
 	WriteHandler    *batchWriter
 	ActivityWriter  *activityWriter
@@ -127,6 +129,13 @@ func WithSettingsCache(settingsCache *settings.RedisSettingsCache) Option {
 	}
 }
 
+// WithSessionHints shares interactive session polling hints with the API.
+func WithSessionHints(hints *cache.SessionHints) Option {
+	return func(h *HandlersTLS) {
+		h.SessionHints = hints
+	}
+}
+
 // WithNodes to pass value as option
 func WithNodes(nodes *nodes.NodeManager) Option {
 	return func(h *HandlersTLS) {
@@ -286,6 +295,9 @@ func (h *HandlersTLS) allowsAcceleratedQueries(queryAccelerated bool) bool {
 }
 
 func (h *HandlersTLS) shouldAccelerateQueryRead(node nodes.OsqueryNode, queryAccelerated bool) bool {
+	if h.OsqueryValues == nil || !h.OsqueryValues.Accelerated {
+		return false
+	}
 	if queryAccelerated {
 		return true
 	}
@@ -296,34 +308,32 @@ func (h *HandlersTLS) hasActiveConsoleSession(node nodes.OsqueryNode) bool {
 	if h.OsqueryValues == nil || !h.OsqueryValues.Console {
 		return false
 	}
-	if node.ID == 0 || node.UUID == "" || node.EnvironmentID == 0 || h.Queries == nil || h.Queries.DB == nil {
-		return false
-	}
-	var count int64
-	if err := h.Queries.DB.Model(&console.Session{}).
-		Where("node_id = ? AND node_uuid = ? AND environment_id = ? AND active = ? AND updated_at >= ?", node.ID, node.UUID, node.EnvironmentID, true, time.Now().Add(-consoleSessionFreshness)).
-		Count(&count).Error; err != nil {
-		log.Debug().Err(err).Msg("error checking active console session for accelerated query read")
-		return false
-	}
-	return count > 0
+	return h.hasActiveSession(node, "console", &console.Session{})
 }
 
 func (h *HandlersTLS) hasActiveFileExplorerSession(node nodes.OsqueryNode) bool {
 	if h.OsqueryValues == nil || !h.OsqueryValues.FileExplorer {
 		return false
 	}
+	return h.hasActiveSession(node, "fileexplorer", &fileexplorer.Session{})
+}
+
+func (h *HandlersTLS) hasActiveSession(node nodes.OsqueryNode, kind string, model any) bool {
 	if node.ID == 0 || node.UUID == "" || node.EnvironmentID == 0 || h.Queries == nil || h.Queries.DB == nil {
 		return false
 	}
-	var count int64
-	if err := h.Queries.DB.Model(&fileexplorer.Session{}).
-		Where("node_id = ? AND node_uuid = ? AND environment_id = ? AND active = ? AND updated_at >= ?", node.ID, node.UUID, node.EnvironmentID, true, time.Now().Add(-consoleSessionFreshness)).
-		Count(&count).Error; err != nil {
-		log.Debug().Err(err).Msg("error checking active file explorer session for accelerated query read")
+	active, err := h.SessionHints.Active(context.Background(), kind, node.EnvironmentID, node.ID, node.UUID, func() (time.Time, error) {
+		var session struct{ UpdatedAt time.Time }
+		err := h.Queries.DB.Model(model).Select("updated_at").
+			Where("node_id = ? AND node_uuid = ? AND environment_id = ? AND active = ? AND updated_at >= ?", node.ID, node.UUID, node.EnvironmentID, true, time.Now().Add(-consoleSessionFreshness)).
+			Order("updated_at DESC").Limit(1).Find(&session).Error
+		return session.UpdatedAt, err
+	})
+	if err != nil {
+		log.Debug().Err(err).Str("kind", kind).Msg("error checking active session for accelerated query read")
 		return false
 	}
-	return count > 0
+	return active
 }
 
 func (h *HandlersTLS) acceleratedSeconds(ctx context.Context) int {

cmd/tls/handlers/post.go

@@ -237,9 +237,6 @@ func (h *HandlersTLS) ConfigHandler(w http.ResponseWriter, r *http.Request) {
 		}
 		// Node and environment match, so we can proceed to update the node
 		ip := utils.GetIP(r)
-		if ip == node.IPAddress {
-			ip = ""
-		}
 		h.WriteHandler.addEvent(lastSeenUpdate{NodeID: node.ID, IP: ip, SeenAt: time.Now()})
 		log.Debug().Msgf("node-uuid: %s with nodeid %d added to batch writer for config update", node.UUID, node.ID)
 		h.recordActivity(env.UUID, node.UUID, activity.EventConfig)
@@ -451,21 +448,18 @@ func (h *HandlersTLS) QueryReadHandler(w http.ResponseWriter, r *http.Request) {
 		// Record ingested data
 		requestSize.WithLabelValues(string(env.UUID), "QueryRead").Observe(float64(len(body)))
 		log.Debug().Msgf("node UUID: %s in %s environment ingested %d bytes for QueryReadHandler endpoint", node.UUID, env.Name, len(body))
+		// Authentication succeeded even if dispatch SQL is temporarily unavailable.
+		h.WriteHandler.addEvent(lastSeenUpdate{NodeID: node.ID, IP: utils.GetIP(r), SeenAt: time.Now()})
+		h.recordActivity(env.UUID, node.UUID, activity.EventQueryRead)
 		// Get queries and update node
 		nodeInvalid = false
 		qs, accelerate, err = h.Queries.NodeQueries(node)
 		if err != nil {
 			log.Err(err).Msg("error getting queries from db")
+			utils.HTTPResponse(w, "", http.StatusServiceUnavailable, []byte(""))
+			return
 		}
 		accelerate = h.shouldAccelerateQueryRead(node, accelerate)
-		// Refresh node last seen
-		ip := utils.GetIP(r)
-		if ip == node.IPAddress {
-			ip = ""
-		}
-		h.WriteHandler.addEvent(lastSeenUpdate{NodeID: node.ID, IP: ip, SeenAt: time.Now()})
-		log.Debug().Msgf("node-uuid: %s with nodeid %d added to batch writer for query read update", node.UUID, node.ID)
-		h.recordActivity(env.UUID, node.UUID, activity.EventQueryRead)
 	} else {
 		log.Err(nodeErr).Msg("GetByKey")
 		nodeInvalid = true
@@ -566,9 +560,6 @@ func (h *HandlersTLS) QueryWriteHandler(w http.ResponseWriter, r *http.Request)
 		}
 		// Refresh node last seen
 		ip := utils.GetIP(r)
-		if ip == node.IPAddress {
-			ip = ""
-		}
 		h.WriteHandler.addEvent(lastSeenUpdate{NodeID: node.ID, IP: ip, SeenAt: time.Now()})
 		// Process submitted results and mark query as processed
 		h.recordActivity(env.UUID, node.UUID, activity.EventQueryWrite)
@@ -814,9 +805,6 @@ func (h *HandlersTLS) CarveInitHandler(w http.ResponseWriter, r *http.Request) {
 		}
 		// Refresh last seen
 		ip := utils.GetIP(r)
-		if ip == node.IPAddress {
-			ip = ""
-		}
 		h.WriteHandler.addEvent(lastSeenUpdate{NodeID: node.ID, IP: ip, SeenAt: time.Now()})
 	}
 	// Prepare response