Skip to content

Commit e9703c2

Browse files
fix(vector): let kit drop blank embedding inputs instead of classifying rejections (#1286)
Supersedes #1276, whose commit this branch keeps and then replaces. #1276 taught the embeddings permanence classifier to recognize a provider's wording for "empty or whitespace-only input" so one blank document would be skip-stamped rather than wedging the fill. Matching provider prose is a losing game: every endpoint words it differently, and the match has to stay narrow enough not to catch an auth or route failure, since skip-stamping one of those would silently mark a whole corpus embedded-with-no-vectors. kit v0.13.1 removes the need for it. `kitvec.Split` omits blank windows and `kitvec.EncodeBatched` refuses blank chunks before any HTTP call — counting whitespace, invisible formatting runes, and control characters as blank — so a blank document is stamped for the generation with no vectors and never becomes a request. The body heuristic is gone; `isPermanentEncodeError` instead recognizes kit's `ErrEmptyEmbeddingInput` sentinel, the same rejection expressed structurally. The upgrade also changes two things this package relied on, both fixed here. A positive `Batch.BatchSize` (the config default is 32) now packs chunks from several documents into one encode call, so a permanent per-input rejection arrives with no attribution and kit aborts the fill unless `ShouldIsolateBatchError` permits document-slice diagnosis. Without that hook the poison-document wedge `OnEncodeError` exists to prevent comes straight back for any real build; the existing regression test missed it only because it left `BatchSize` at zero, which selects kit's legacy per-document path. Only permanent errors authorize the extra probe calls, so transient failures still abort untouched. Blank windows are dropped but still numbered, so a document with an all-whitespace window is now legitimately stored with a gap in its chunk indexes. Two places assumed those indexes were dense: search resolved snippets by slice position, which returned nothing for chunks after a gap, and the repair scan condemned such a document as structurally incomplete, which would re-embed it on every repair run indefinitely. Reviewers should look at `ShouldIsolateBatchError` in `internal/vector/build.go` — it is the one change that alters fill control flow — and at the `state.expected` rewrite in `internal/vector/repair.go`, which decides whether a document is considered healthy. <sup>generated by a clanker</sup> Co-authored-by: Marius van Niekerk <mariusvniekerk@users.noreply.github.com>
1 parent 33c9695 commit e9703c2

7 files changed

Lines changed: 254 additions & 19 deletions

File tree

go.mod

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ require (
2727
github.com/testcontainers/testcontainers-go/modules/postgres v0.43.0
2828
github.com/tidwall/gjson v1.19.0
2929
go.kenn.io/docbank v0.11.0
30-
go.kenn.io/kit v0.11.0
30+
go.kenn.io/kit v0.13.1
3131
golang.org/x/mod v0.37.0
3232
golang.org/x/perf v0.0.0-20260615155930-9e4b9ddef5b6
3333
golang.org/x/sync v0.21.0

go.sum

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -285,6 +285,10 @@ go.kenn.io/docbank v0.11.0 h1:MLv0CxCWyk5kM5zX3YALquVxi7sjAjrcMD/XNQkuw6Y=
285285
go.kenn.io/docbank v0.11.0/go.mod h1:4hym+ONhsG8epaBgTXA1oC/PnF60Z3CyvcRs44qKhb8=
286286
go.kenn.io/kit v0.11.0 h1:OdEaI8i3R7M0OTptrP2Osu+WJSg+lTClmHalRgn/T/U=
287287
go.kenn.io/kit v0.11.0/go.mod h1:dComZhFNb4LR+Tj4ZD0slEDMkPJk0gGd8q/HEHXTGSM=
288+
go.kenn.io/kit v0.13.0 h1:N4/KvR1xnM2o97q2CHnBWAt1uXcNqxTFGT0dbkiloEA=
289+
go.kenn.io/kit v0.13.0/go.mod h1:dComZhFNb4LR+Tj4ZD0slEDMkPJk0gGd8q/HEHXTGSM=
290+
go.kenn.io/kit v0.13.1 h1:KQxCS2GMczrrWhZgxjDwD480hB4nXgtw7jfBuJujhWg=
291+
go.kenn.io/kit v0.13.1/go.mod h1:dComZhFNb4LR+Tj4ZD0slEDMkPJk0gGd8q/HEHXTGSM=
288292
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
289293
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
290294
go.opentelemetry.io/contrib/bridges/prometheus v0.69.0 h1:saQoWg5845Q8TojpqeVStS7zGwVZ6bc5W2PJavTPiBM=

internal/vector/build.go

Lines changed: 28 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -186,10 +186,18 @@ func (ix *Index) Build(
186186
spec: ix.spec,
187187
}
188188
fillStats, fillErr := kitvec.Fill[string, string](ctx, fillStore, target, wrapped, kitvec.FillOptions[string]{
189-
Split: ix.split,
190-
Batch: kitvec.BatchOptions{BatchSize: o.BatchSize, Concurrency: 1},
191-
Concurrency: o.Concurrency,
192-
OnEncodeError: skipPermanentEncodeError,
189+
Split: ix.split,
190+
Batch: kitvec.BatchOptions{BatchSize: o.BatchSize, Concurrency: 1},
191+
Concurrency: o.Concurrency,
192+
// A positive BatchSize packs chunks from several documents into one
193+
// encode call, so a permanent rejection arrives without knowing which
194+
// document caused it. Permitting isolation lets kit re-encode each
195+
// document's slice alone and attribute the failure; without it the
196+
// whole fill aborts and one poison document wedges every later build,
197+
// which is what OnEncodeError exists to prevent. Transient failures
198+
// still abort untouched: they are not worth N extra probe calls.
199+
ShouldIsolateBatchError: isPermanentEncodeError,
200+
OnEncodeError: skipPermanentEncodeError,
193201
})
194202
finish()
195203
result.Fill = fillStats
@@ -320,13 +328,12 @@ func validatingEncoder(enc kitvec.EncodeFunc) kitvec.EncodeFunc {
320328

321329
// skipPermanentEncodeError implements kitvec.FillOptions.OnEncodeError: a
322330
// document the embeddings endpoint permanently rejects for input-specific
323-
// reasons (e.g. a token-window overflow, whitespace-only content some servers
324-
// refuse, or a content-policy rejection) is skipped — kit stamps it for the
325-
// generation with no vectors so it stops being pending — instead of aborting
326-
// the whole fill. Without this, one poison document would wedge every future
327-
// build at the same doc_key-ordered scan position: later documents would never
328-
// embed, a first build would never reach Missing==0, and auto-activation would
329-
// never fire.
331+
// reasons (e.g. a token-window overflow or a content-policy rejection) is
332+
// skipped — kit stamps it for the generation with no vectors so it stops being
333+
// pending — instead of aborting the whole fill. Without this, one poison
334+
// document would wedge every future build at the same doc_key-ordered scan
335+
// position: later documents would never embed, a first build would never reach
336+
// Missing==0, and auto-activation would never fire.
330337
//
331338
// Every other failure (5xx, network, timeout, 429 rate-limiting, auth, route,
332339
// model, media-type, or other config/API failures) still aborts the fill, since
@@ -341,7 +348,17 @@ func skipPermanentEncodeError(doc string, err error) bool {
341348
return true
342349
}
343350

351+
// isPermanentEncodeError reports whether err rejects one specific input in a
352+
// way retrying can never fix. kitvec.ErrEmptyEmbeddingInput is kit's own
353+
// pre-flight refusal of blank chunk text; it is raised before any HTTP call
354+
// and replaces sniffing each provider's wording for the same rejection.
355+
// Ordinary fills never trigger it — kitvec.Split drops blank windows, so a
356+
// blank document is stamped with no vectors — but a chunk that reaches an
357+
// encode call blank is still permanently unembeddable, not a transient fault.
344358
func isPermanentEncodeError(err error) bool {
359+
if errors.Is(err, kitvec.ErrEmptyEmbeddingInput) {
360+
return true
361+
}
345362
var statusErr *HTTPStatusError
346363
return errors.As(err, &statusErr) && statusErr != nil && statusErr.Permanent()
347364
}

internal/vector/encoder.go

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,13 @@ func (e *HTTPStatusError) Permanent() bool {
128128
// media-type failure, and skip-stamping those would silently mark the whole
129129
// corpus embedded-with-no-vectors — so a size word must pair with an input
130130
// word, and "content" must pair with "policy".
131+
//
132+
// Blank inputs are deliberately absent: kitvec.Split drops blank windows and
133+
// kitvec.EncodeBatched refuses blank chunks outright — counting whitespace,
134+
// invisible formatting runes, and control characters as blank — so a blank
135+
// document is stamped without vectors and never becomes a request. That
136+
// removes the need to recognize each provider's phrasing for the rejection
137+
// (see isPermanentEncodeError for the structured signal kit raises instead).
131138
func hasDocumentSpecificEmbeddingError(body string) bool {
132139
body = strings.ToLower(body)
133140
if strings.Contains(body, "content") && strings.Contains(body, "policy") {

internal/vector/repair.go

Lines changed: 22 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -132,6 +132,17 @@ SELECT c.doc_key
132132
return documents, nil
133133
}
134134

135+
// chunkIndexes returns the chunk indexes content splits into, in order. It is
136+
// not always 0..n-1: kitvec.Split numbers by window and omits blank ones.
137+
func chunkIndexes(content string, split kitvec.SplitOptions) []int {
138+
chunks := kitvec.Split(content, split)
139+
indexes := make([]int, len(chunks))
140+
for i, chunk := range chunks {
141+
indexes[i] = chunk.Index
142+
}
143+
return indexes
144+
}
145+
135146
func (ix *Index) scanInvalidRepairDocuments(
136147
ctx context.Context, ordinal int64, dimension int, vecTable string, documents []string,
137148
) ([]string, error) {
@@ -144,8 +155,15 @@ func (ix *Index) scanInvalidRepairDocuments(
144155
args = append(args, docKey)
145156
}
146157
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(documents)), ",")
158+
// expected is the chunk indexes this content splits into, in order, not a
159+
// count: kitvec.Split omits blank windows while numbering by window, so a
160+
// document with an all-whitespace window is correctly stored with a gap in
161+
// its chunk indexes. Comparing each stored index against the expected one
162+
// keeps that document healthy while still catching genuine gaps, whereas
163+
// an "index equals position" rule would condemn it on every repair scan
164+
// and re-embed it forever.
147165
type documentState struct {
148-
expected int
166+
expected []int
149167
seen int
150168
invalid bool
151169
}
@@ -167,7 +185,7 @@ SELECT d.doc_key, d.content
167185
contentRows.Close()
168186
return nil, fmt.Errorf("scan repair document content: %w", err)
169187
}
170-
states[docKey] = &documentState{expected: len(kitvec.Split(content, ix.split))}
188+
states[docKey] = &documentState{expected: chunkIndexes(content, ix.split)}
171189
}
172190
if err := contentRows.Err(); err != nil {
173191
contentRows.Close()
@@ -203,7 +221,7 @@ SELECT c.doc_key, c.chunk_index, v.embedding
203221
if state == nil {
204222
continue
205223
}
206-
if state.seen >= state.expected || state.seen != chunkIndex {
224+
if state.seen >= len(state.expected) || state.expected[state.seen] != chunkIndex {
207225
state.invalid = true
208226
}
209227
state.seen++
@@ -217,7 +235,7 @@ SELECT c.doc_key, c.chunk_index, v.embedding
217235
var affected []string
218236
for _, docKey := range documents {
219237
state := states[docKey]
220-
if state != nil && (state.invalid || state.seen != state.expected) {
238+
if state != nil && (state.invalid || state.seen != len(state.expected)) {
221239
affected = append(affected, docKey)
222240
}
223241
}

internal/vector/search.go

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -393,12 +393,21 @@ func (ix *Index) snippet(content string, chunkIndex int) string {
393393
// chunkSnippet is the snippet method's package-level body, shared with
394394
// DocAnchor for backends (the PG vector searcher) that resolve chunk hits
395395
// against a document loaded outside an Index.
396+
//
397+
// Chunk.Index counts windows, not surviving chunks: kitvec.Split omits blank
398+
// windows, so a document with an all-whitespace window stores chunk indexes
399+
// with a gap in them. Match on Index rather than slice position or such a hit
400+
// resolves to a neighboring chunk's text, or to none at all.
396401
func chunkSnippet(content string, chunkIndex int, split kitvec.SplitOptions) string {
397-
chunks := kitvec.Split(content, split)
398-
if chunkIndex < 0 || chunkIndex >= len(chunks) {
402+
if chunkIndex < 0 {
399403
return ""
400404
}
401-
return truncateRunes(chunks[chunkIndex].Text, snippetMaxRunes)
405+
for _, chunk := range kitvec.Split(content, split) {
406+
if chunk.Index == chunkIndex {
407+
return truncateRunes(chunk.Text, snippetMaxRunes)
408+
}
409+
}
410+
return ""
402411
}
403412

404413
// truncateRunes truncates s to at most maxRunes runes, appending an
Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
package vector
2+
3+
import (
4+
"context"
5+
"fmt"
6+
"slices"
7+
"strings"
8+
"testing"
9+
10+
"github.com/stretchr/testify/assert"
11+
"github.com/stretchr/testify/require"
12+
kitvec "go.kenn.io/kit/vector"
13+
)
14+
15+
// blankWindowContent is long enough that its middle 4000-rune window (the
16+
// test index's MaxRunes, stride 3400 after the 15% overlap) holds nothing but
17+
// spaces, so kitvec.Split drops that window and numbers the surviving chunks
18+
// 0 and 2. Everything that re-derives chunks from content has to cope with
19+
// that gap.
20+
func blankWindowContent() string {
21+
return strings.Repeat("a", 100) + strings.Repeat(" ", 7500) + strings.Repeat("b", 100)
22+
}
23+
24+
// recordingEncoder returns a unit-vector encoder that appends every text it is
25+
// asked to embed to seen.
26+
func recordingEncoder(seen *[]string) kitvec.EncodeFunc {
27+
return func(_ context.Context, texts []string) ([][]float32, error) {
28+
*seen = append(*seen, texts...)
29+
out := make([][]float32, len(texts))
30+
for i := range texts {
31+
out[i] = []float32{1, 0, 0}
32+
}
33+
return out, nil
34+
}
35+
}
36+
37+
// TestBuildStampsWhitespaceOnlyDocumentWithoutEmbeddingIt covers the whole
38+
// point of the kit 0.13 upgrade: a document whose content is only whitespace
39+
// is stamped for the generation with no vectors and never becomes an
40+
// embeddings request, so no provider ever gets the chance to reject it. The
41+
// build still completes and auto-activates.
42+
func TestBuildStampsWhitespaceOnlyDocumentWithoutEmbeddingIt(t *testing.T) {
43+
ix := openTestIndex(t)
44+
ctx := context.Background()
45+
src := &fakeUnitSource{rows: []fakeUnit{
46+
{unit: userDoc("s1", "", 0, " \n\t   "), endedAt: "2024-01-01T00:00:00Z"},
47+
{unit: userDoc("s1", "", 1, "real content"), endedAt: "2024-01-01T00:00:01Z"},
48+
}}
49+
50+
var seen []string
51+
result, err := ix.Build(ctx, src, recordingEncoder(&seen), fakeGeneration("fake-model"),
52+
BuildOptions{BatchSize: 32})
53+
require.NoError(t, err)
54+
assert.Equal(t, []string{"real content"}, seen,
55+
"whitespace-only content must never reach the embeddings endpoint")
56+
assert.True(t, result.Activated,
57+
"a stamped-without-vectors blank document still counts as covered")
58+
59+
var stamps int
60+
require.NoError(t, ix.db.QueryRow(
61+
`SELECT COUNT(*) FROM message_vectors_stamps`).Scan(&stamps))
62+
assert.Equal(t, 2, stamps, "both documents are stamped")
63+
64+
var chunks int
65+
require.NoError(t, ix.db.QueryRow(
66+
`SELECT COUNT(*) FROM message_vectors_chunks`).Scan(&chunks))
67+
assert.Equal(t, 1, chunks, "only the non-blank document has a chunk")
68+
}
69+
70+
// TestEmptyEmbeddingInputIsPermanent pins the structured signal that replaced
71+
// sniffing provider error bodies for whitespace wording: kit refuses a blank
72+
// chunk before any HTTP call, and that refusal is a permanent per-document
73+
// rejection, so a fill stamp-skips it instead of retrying it forever.
74+
func TestEmptyEmbeddingInputIsPermanent(t *testing.T) {
75+
err := fmt.Errorf("encode chunk 0: %w", kitvec.ErrEmptyEmbeddingInput)
76+
assert.True(t, isPermanentEncodeError(err))
77+
}
78+
79+
// TestBuildSkipsPermanentlyRejectedDocumentSharingABatch is the cross-document
80+
// batching counterpart to TestBuildSkipsPermanentlyRejectedDocumentAndContinues.
81+
// A configured batch_size (production always sets one) packs chunks from
82+
// several documents into one encode call, so a permanent rejection arrives
83+
// with no attribution. The build must still isolate the offending document and
84+
// skip only it; aborting would wedge every later build at the same document.
85+
func TestBuildSkipsPermanentlyRejectedDocumentSharingABatch(t *testing.T) {
86+
ix := openTestIndex(t)
87+
ctx := context.Background()
88+
src := &fakeUnitSource{rows: []fakeUnit{
89+
{unit: userDoc("s1", "", 0, "one"), endedAt: "2024-01-01T00:00:00Z"},
90+
{unit: userDoc("s1", "", 1, "poison"), endedAt: "2024-01-01T00:00:01Z"},
91+
{unit: userDoc("s1", "", 2, "three"), endedAt: "2024-01-01T00:00:02Z"},
92+
}}
93+
94+
rejectPoison := func(_ context.Context, texts []string) ([][]float32, error) {
95+
if slices.Contains(texts, "poison") {
96+
return nil, &HTTPStatusError{
97+
Status: 400, Body: "input exceeds maximum context length",
98+
}
99+
}
100+
out := make([][]float32, len(texts))
101+
for i := range texts {
102+
out[i] = []float32{1, 0, 0}
103+
}
104+
return out, nil
105+
}
106+
107+
result, err := ix.Build(ctx, src, rejectPoison, fakeGeneration("fake-model"),
108+
BuildOptions{BatchSize: 32})
109+
require.NoError(t, err,
110+
"one poison document in a shared batch must not abort the whole build")
111+
assert.Equal(t, 2, result.Fill.Documents, "the two good documents still embed")
112+
assert.Equal(t, 1, result.Fill.Skipped, "only the poison document is skipped")
113+
assert.True(t, result.Activated)
114+
}
115+
116+
// TestBuildTransientBatchErrorStillAborts guards the other side of batch
117+
// isolation: a 5xx applies to the call, not to one input, so it must abort
118+
// without probing each document slice separately.
119+
func TestBuildTransientBatchErrorStillAborts(t *testing.T) {
120+
ix := openTestIndex(t)
121+
ctx := context.Background()
122+
src := &fakeUnitSource{rows: []fakeUnit{
123+
{unit: userDoc("s1", "", 0, "one"), endedAt: "2024-01-01T00:00:00Z"},
124+
{unit: userDoc("s1", "", 1, "two"), endedAt: "2024-01-01T00:00:01Z"},
125+
}}
126+
127+
var calls int
128+
failing := func(_ context.Context, _ []string) ([][]float32, error) {
129+
calls++
130+
return nil, &HTTPStatusError{Status: 503, Body: "upstream unavailable"}
131+
}
132+
133+
result, err := ix.Build(ctx, src, failing, fakeGeneration("fake-model"),
134+
BuildOptions{BatchSize: 32})
135+
require.Error(t, err)
136+
assert.Zero(t, result.Fill.Skipped, "a transient failure must never skip-stamp")
137+
assert.Equal(t, 1, calls, "a transient failure must not trigger per-document probes")
138+
}
139+
140+
// TestChunkSnippetResolvesIndexAcrossADroppedWindow covers a search hit on a
141+
// document with a blank window: its chunk indexes have a gap, so resolving a
142+
// snippet by slice position would return the wrong chunk's text (or none).
143+
func TestChunkSnippetResolvesIndexAcrossADroppedWindow(t *testing.T) {
144+
split := kitvec.SplitOptions{MaxRunes: 10}
145+
content := strings.Repeat("a", 10) + strings.Repeat(" ", 10) + strings.Repeat("b", 10)
146+
chunks := kitvec.Split(content, split)
147+
require.Len(t, chunks, 2, "the all-whitespace middle window is dropped")
148+
require.Equal(t, []int{0, 2}, []int{chunks[0].Index, chunks[1].Index},
149+
"the surviving chunks keep their window numbers")
150+
151+
assert.Equal(t, strings.Repeat("a", 10), chunkSnippet(content, 0, split))
152+
assert.Equal(t, strings.Repeat("b", 10), chunkSnippet(content, 2, split),
153+
"the second stored chunk resolves by its index, not its slice position")
154+
assert.Empty(t, chunkSnippet(content, 1, split),
155+
"the dropped window has no snippet")
156+
}
157+
158+
// TestRepairKeepsDocumentWithADroppedWindow covers the repair scan's view of
159+
// the same document: chunk indexes 0 and 2 are exactly what Split asks for, so
160+
// repair must leave the document alone. Treating the gap as a missing chunk
161+
// would re-embed the document on every repair run, forever.
162+
func TestRepairKeepsDocumentWithADroppedWindow(t *testing.T) {
163+
ix := openTestIndex(t)
164+
ctx := context.Background()
165+
src := &fakeUnitSource{rows: []fakeUnit{
166+
{unit: userDoc("s1", "", 0, blankWindowContent()), endedAt: "2024-01-01T00:00:00Z"},
167+
}}
168+
gen := fakeGeneration("fake-model")
169+
170+
var seen []string
171+
built, err := ix.Build(ctx, src, recordingEncoder(&seen), gen, BuildOptions{BatchSize: 32})
172+
require.NoError(t, err)
173+
require.Equal(t, 2, built.Fill.Chunks)
174+
175+
repaired, err := ix.Build(ctx, src, recordingEncoder(&seen), gen,
176+
BuildOptions{BatchSize: 32, RepairInvalid: true})
177+
require.NoError(t, err)
178+
assert.Zero(t, repaired.Repair.Documents,
179+
"a document whose chunk indexes skip a blank window is healthy")
180+
}

0 commit comments

Comments
 (0)