fix(collection,query): harden take/drop count-overflow family

src/ops/collection.c

@@ -1629,7 +1629,7 @@ ray_t* ray_take_fn(ray_t* vec, ray_t* n_obj) {
      * floats up front: as_i64(f64) reads the bit pattern and would cause
      * e.g. (take 1.0 2.0) to attempt a 4.6-quintillion-element allocation
      * and surface as "oom" — misleading for what is really a type error. */
-    if (ray_is_atom(n_obj) && n_obj->type == -RAY_F64)
+    if (ray_is_atom(n_obj) && (n_obj->type == -RAY_F64 || n_obj->type == -RAY_F32))
         return ray_error("type", "take: count must be an integer, got %s", ray_type_name(n_obj->type));
     /* Range take: (take collection [start amount]) — slice from start for amount elements */
     if (ray_is_vec(n_obj) && n_obj->type == RAY_I64 && ray_len(n_obj) == 2) {
@@ -1662,8 +1662,9 @@ ray_t* ray_take_fn(ray_t* vec, ray_t* n_obj) {
             if (start < 0) start = slen + start;
             if (start < 0) start = 0;
             if (start >= slen) return ray_str("", 0);
-            int64_t end = start + amount;
-            if (end > slen) end = slen;
+            /* amount can be up to INT64_MAX, so `start + amount` overflows;
+             * take min(amount, slen - start) instead (0 <= start < slen). */
+            int64_t end = (amount < slen - start) ? start + amount : slen;
             return ray_str(s + start, (size_t)(end - start));
         }
 
@@ -1680,8 +1681,9 @@ ray_t* ray_take_fn(ray_t* vec, ray_t* n_obj) {
                     col_propagate_str_pool(empty, vec);
                 return empty;
             }
-            int64_t end = start + amount;
-            if (end > len) end = len;
+            /* Overflow-safe clamp: amount may be up to INT64_MAX, so avoid
+             * computing start+amount unless it is known to fit (0 <= start < len). */
+            int64_t end = (amount < len - start) ? start + amount : len;
             int64_t count = end - start;
             int8_t vtype = vec->type;
             int esz = ray_sym_elem_size(vtype, vec->attrs);
@@ -1716,8 +1718,9 @@ ray_t* ray_take_fn(ray_t* vec, ray_t* n_obj) {
             int64_t len = keys ? keys->len : 0;
             if (start < 0) start = len + start;
             if (start < 0) start = 0;
-            int64_t end = start + amount;
-            if (end > len) end = len;
+            /* Overflow-safe clamp (amount up to INT64_MAX); keep the end<start
+             * guard because the dict branch does not early-return start>=len. */
+            int64_t end = (amount < len - start) ? start + amount : len;
             if (end < start) end = start;
             int64_t count = end - start;
 
@@ -1754,8 +1757,7 @@ ray_t* ray_take_fn(ray_t* vec, ray_t* n_obj) {
                 result->len = 0;
                 return result;
             }
-            int64_t end = start + amount;
-            if (end > len) end = len;
+            int64_t end = (amount < len - start) ? start + amount : len;
             int64_t count = end - start;
             ray_t** elems = (ray_t**)ray_data(vec);
             ray_t* result = ray_alloc(count * sizeof(ray_t*));
@@ -1773,11 +1775,13 @@ ray_t* ray_take_fn(ray_t* vec, ray_t* n_obj) {
         return ray_error("type", "take: range take unsupported for %s", ray_type_name(vec->type));
     }
     /* Every scalar-count branch below turns the count into a magnitude with
-     * `n < 0 ? -n : n`; `-INT64_MIN` is signed-overflow UB and |INT64_MIN| is
-     * unallocatable anyway, so reject an INT64_MIN count (also the i64 null
-     * sentinel) once here, up front, for all of them. */
-    if (ray_is_atom(n_obj) && is_numeric(n_obj) && as_i64(n_obj) == INT64_MIN)
-        return ray_error("range", "take: count magnitude out of range");
+     * `n < 0 ? -n : n`, and `-INT64_MIN` is signed-overflow UB.  INT64_MIN is
+     * also the i64 null sentinel (0N), so reject it once here as a type error
+     * — symmetric with drop and the window verbs — rather than letting a null
+     * count through.  Float counts are already rejected above, so is_numeric
+     * here only admits integer atoms (and implies ray_is_atom). */
+    if (is_numeric(n_obj) && as_i64(n_obj) == INT64_MIN)
+        return ray_error("type", "take: count is null or out of range");
     /* Char take: (take 'a' n) → string of n copies of char */
     if (ray_is_atom(vec) && vec->type == -RAY_STR && ray_str_len(vec) == 1 && ray_is_atom(n_obj) && is_numeric(n_obj)) {
         int64_t n = as_i64(n_obj);
@@ -2015,16 +2019,18 @@ ray_t* ray_drop_fn(ray_t* vec, ray_t* n_obj) {
         return ray_error("type", "drop: expected a collection, got %s", vec ? ray_type_name(vec->type) : "null");
 
     int64_t n = as_i64(n_obj);
+    /* INT64_MIN is the i64 null sentinel (0N) and `-INT64_MIN` is signed-
+     * overflow UB; reject it as a type error, symmetric with take, rather
+     * than silently dropping the whole collection. */
+    if (n == INT64_MIN)
+        return ray_error("type", "drop: count is null or out of range");
     int64_t start = 0;
     int64_t amount = len;
     if (n >= 0) {
         start = n < len ? n : len;
         amount = len - start;
     } else {
-        /* `-INT64_MIN` is signed-overflow UB; a drop-from-end of that
-         * magnitude removes the whole collection (|n| >= len), so treat it as
-         * cut == len (amount 0) rather than negating. */
-        int64_t cut = (n == INT64_MIN) ? len : -n;
+        int64_t cut = -n;   /* n != INT64_MIN here, so negation is safe */
         amount = cut < len ? len - cut : 0;
     }
     return collection_slice(vec, start, amount);

src/ops/query.c

@@ -623,6 +623,12 @@ static ray_t* apply_sort_take(ray_t* result, ray_t** dict_elems, int64_t dict_n,
         if (ray_is_atom(tv) && (tv->type == -RAY_I64 || tv->type == -RAY_I32)) {
             int64_t atom_n = (tv->type == -RAY_I64) ? tv->i64 : tv->i32;
             ray_release(tv);
+            /* INT64_MIN is the i64 null sentinel (0N); negating it below is
+             * signed-overflow UB.  Reject it up front, matching ray_take_fn. */
+            if (atom_n == INT64_MIN) {
+                ray_release(result);
+                return ray_error("type", "select: take: count is null or out of range");
+            }
 
             int64_t nrows = (result->type == RAY_TABLE)
                           ? ray_table_nrows(result)
@@ -862,6 +868,11 @@ static ray_t* apply_sort_take(ray_t* result, ray_t** dict_elems, int64_t dict_n,
         int64_t nrows = (sorted->type == RAY_TABLE)
                       ? ray_table_nrows(sorted)
                       : (ray_is_vec(sorted) ? sorted->len : 0);
+        /* INT64_MIN (0N) negated below is signed-overflow UB — reject it. */
+        if (atom_n == INT64_MIN) {
+            ray_release(sorted);
+            return ray_error("type", "select: take: count is null or out of range");
+        }
         int64_t start, amount;
         if (atom_n >= 0) {
             start  = 0;
@@ -9927,6 +9938,13 @@ ray_t* ray_select(ray_t** args, int64_t n) {
         if (!tv || RAY_IS_ERR(tv)) { ray_graph_free(g); ray_release(tbl); scratch_free(sel_slots_hdr); DICT_VIEW_CLOSE(dv); return tv ? tv : ray_error("domain", "select: failed to evaluate `take:`"); }
         if (ray_is_atom(tv) && (tv->type == -RAY_I64 || tv->type == -RAY_I32)) {
             int64_t n_take = (tv->type == -RAY_I64) ? tv->i64 : tv->i32;
+            /* INT64_MIN (0N) negated for a tail pushdown is signed-overflow UB;
+             * reject it before it reaches ray_tail / the TAIL kernel. */
+            if (n_take == INT64_MIN) {
+                ray_release(tv);
+                ray_graph_free(g); ray_release(tbl); scratch_free(sel_slots_hdr); DICT_VIEW_CLOSE(dv);
+                return ray_error("type", "select: take: count is null or out of range");
+            }
             if (group_take_push) {
                 take_pre.kind = TAKE_PRE_ATOM;
                 take_pre.a = n_take;

test/rfl/collection/drop_cut_rotate_cross.rfl

@@ -4,9 +4,10 @@
 (drop [1 2 3 4 5] -2) -- [1 2 3]
 (drop [1 2 3] 10) -- []
 (drop [1 2 3] -10) -- []
-;; INT64_MIN count: `-INT64_MIN` is signed-overflow UB (was tripped in
-;; ray_drop_fn); |n| that large drops the whole collection.
-(drop [1 2 3] -9223372036854775808) -- []
+;; INT64_MIN count is the i64 null sentinel (0N), and `-INT64_MIN` is signed-
+;; overflow UB (was tripped in ray_drop_fn).  Reject it as a type error,
+;; symmetric with take, rather than silently dropping the whole collection.
+(drop [1 2 3] -9223372036854775808) !- type
 (drop "abcdef" 2) -- "cdef"
 (drop "abcdef" -2) -- "abcd"
 (drop ['a 'b 'c] 1) -- ['b 'c]

test/rfl/collection/take.rfl

@@ -175,12 +175,21 @@
 (count (take [1 2 3 4 5] [2 0])) -- 0
 ;; range take with start = count → empty
 (count (take [1 2 3 4 5] [5 3])) -- 0
-;; INT64_MIN count: `-INT64_MIN` is signed-overflow UB (was tripped computing
-;; the magnitude in every scalar-count take branch — vector, string, char,
-;; scalar, list).  Its magnitude is unrepresentable as int64, so it is a range
-;; error, not UB followed by a bogus allocation.
-(take [1 2 3] -9223372036854775808) !- range
-(take "hello" -9223372036854775808) !- range
-(take "a" -9223372036854775808) !- range
-(take 42 -9223372036854775808) !- range
-(take (list 1 2 3) -9223372036854775808) !- range
+;; INT64_MIN count is the i64 null sentinel (0N), and `-INT64_MIN` is signed-
+;; overflow UB (was tripped computing the magnitude in every scalar-count take
+;; branch — vector, string, char, scalar, list).  Reject it as a *type* error,
+;; symmetric with drop and the window verbs, rather than negating a null count.
+(take [1 2 3] -9223372036854775808) !- type
+(take "hello" -9223372036854775808) !- type
+(take "a" -9223372036854775808) !- type
+(take 42 -9223372036854775808) !- type
+(take (list 1 2 3) -9223372036854775808) !- type
+;; Float counts are rejected up front as a type error.  The guard also covers
+;; F32 (as_i64 on an f32 would bit-read the storage and mis-fire the INT64_MIN
+;; check), but F32 atoms have no rfl literal / `as` form, so only F64 is pinned.
+(take [1 2 3] 2.5) !- type
+;; Range take: amount up to INT64_MAX must clamp to the available length;
+;; `start + amount` overflowed int64 (UB → bogus count → memcpy) before the fix.
+(take [1 2 3] [1 9223372036854775807]) -- [2 3]
+(take "hello" [1 9223372036854775807]) -- "ello"
+(count (take (list 1 2 3) [1 9223372036854775807])) -- 2

test/rfl/table/select.rfl

@@ -121,6 +121,16 @@
 (count (select {from: trades-15 take: 0})) -- 0
 (count (select {from: trades-15 take: 15})) -- 15
 (sum (at (select {from: trades-15 take: 15}) 'size)) -- 1765
+;; negative take: (from the end), clamped to the row count
+(count (select {from: trades-15 take: -5})) -- 5
+(count (select {from: trades-15 take: -20})) -- 15
+(sum (at (select {from: trades-15 take: -15}) 'size)) -- 1765
+;; INT64_MIN take count is the i64 null sentinel (0N); negating it is signed-
+;; overflow UB.  Every select take: path — DAG tail-pushdown, sorted
+;; apply_sort_take, grouped — must reject it as a type error rather than trap.
+(select {from: trades-15 take: -9223372036854775808}) !- type
+(select {from: trades-15 asc: size take: -9223372036854775808}) !- type
+(select {c: (count size) from: trades-15 by: sym take: -9223372036854775808}) !- type
 
 ;; ── edge cases: all-pass / none-pass / single-row / by-distinct-time
 (count (select {from: trades-15 where: (> size -1)})) -- 15