Fix constant re-ingestion when server provides no revision header
HTTP, S3, and WebDAV sources all check `revision is not None and snapshot.get(uri) == revision` to decide UPSERT vs UNCHANGED. When a server returns no ETag or Last-Modified, revision is None and the condition always fails — every sweep emits UPSERT even though the content hasn't changed. Now emit UNCHANGED when revision is None and the URI is already known (has been ingested before). A first-time discovery with no revision still correctly emits UPSERT.
This commit is contained in:
parent
20634376a3
commit
8deac2fee8
6 changed files with 121 additions and 0 deletions
|
|
@ -144,6 +144,13 @@ class HTTPSource:
|
|||
revision, _ = _extract_revision(head.headers)
|
||||
if revision is not None and snapshot.get(url) == revision:
|
||||
kind = SourceEventKind.UNCHANGED
|
||||
elif revision is None and url in known:
|
||||
# Server provides no revision header (no ETag, no
|
||||
# Last-Modified). We can't detect changes, but the URL was
|
||||
# already ingested — skip rather than re-fetch every sweep. A
|
||||
# real change is only picked up if the server starts returning
|
||||
# revision headers or the operator forces a re-ingest.
|
||||
kind = SourceEventKind.UNCHANGED
|
||||
else:
|
||||
kind = SourceEventKind.UPSERT
|
||||
|
||||
|
|
|
|||
|
|
@ -132,6 +132,8 @@ class S3Source:
|
|||
|
||||
if revision is not None and snapshot.get(uri) == revision:
|
||||
kind = SourceEventKind.UNCHANGED
|
||||
elif revision is None and uri in known:
|
||||
kind = SourceEventKind.UNCHANGED
|
||||
else:
|
||||
kind = SourceEventKind.UPSERT
|
||||
|
||||
|
|
|
|||
|
|
@ -282,6 +282,8 @@ class WebDAVSource:
|
|||
revision = entry.revision
|
||||
if revision is not None and snapshot.get(uri) == revision:
|
||||
kind = SourceEventKind.UNCHANGED
|
||||
elif revision is None and uri in known:
|
||||
kind = SourceEventKind.UNCHANGED
|
||||
else:
|
||||
kind = SourceEventKind.UPSERT
|
||||
yield SourceEvent(
|
||||
|
|
|
|||
|
|
@ -342,3 +342,38 @@ async def test_discover_propagates_non_transport_errors():
|
|||
with pytest.raises(TypeError, match="unexpected bug"):
|
||||
async for _ in src.discover():
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_emits_unchanged_for_known_url_without_revision():
|
||||
"""A server that returns no ETag or Last-Modified should not cause
|
||||
re-ingestion every sweep once the URL has been ingested."""
|
||||
transport = _transport(
|
||||
{("HEAD", "https://example.com/a.md"): httpx.Response(200)}
|
||||
)
|
||||
src = HTTPSource(
|
||||
source_id="x", urls=["https://example.com/a.md"], transport=transport
|
||||
)
|
||||
events = [
|
||||
e
|
||||
async for e in src.discover(
|
||||
known_uris={"https://example.com/a.md"}
|
||||
)
|
||||
]
|
||||
assert len(events) == 1
|
||||
assert events[0].kind is SourceEventKind.UNCHANGED
|
||||
assert events[0].revision is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_emits_upsert_for_unknown_url_without_revision():
|
||||
"""A brand-new URL with no revision should still UPSERT on first sight."""
|
||||
transport = _transport(
|
||||
{("HEAD", "https://example.com/new.md"): httpx.Response(200)}
|
||||
)
|
||||
src = HTTPSource(
|
||||
source_id="x", urls=["https://example.com/new.md"], transport=transport
|
||||
)
|
||||
events = [e async for e in src.discover()]
|
||||
assert len(events) == 1
|
||||
assert events[0].kind is SourceEventKind.UPSERT
|
||||
|
|
|
|||
|
|
@ -213,3 +213,28 @@ async def test_discover_respects_ignore_patterns(fake_s3_listing):
|
|||
)
|
||||
events = [e async for e in src.discover()]
|
||||
assert {e.uri for e in events} == {"s3://bucket/a.md"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_emits_unchanged_for_known_key_without_etag(fake_s3_listing):
|
||||
"""An S3 object with no ETag should not cause re-ingestion every sweep
|
||||
once the key has been ingested."""
|
||||
fake_s3_listing([[{"path": "file.md", "size": 0, "last_modified": None}]])
|
||||
src = S3Source(uri="s3://bucket/", supported_extensions=[".md"])
|
||||
events = [
|
||||
e
|
||||
async for e in src.discover(known_uris={"s3://bucket/file.md"})
|
||||
]
|
||||
non_delete = [e for e in events if e.kind is not SourceEventKind.DELETE]
|
||||
assert len(non_delete) == 1
|
||||
assert non_delete[0].kind is SourceEventKind.UNCHANGED
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_emits_upsert_for_unknown_key_without_etag(fake_s3_listing):
|
||||
"""A brand-new S3 key with no ETag should UPSERT on first sight."""
|
||||
fake_s3_listing([[{"path": "new.md", "size": 0, "last_modified": None}]])
|
||||
src = S3Source(uri="s3://bucket/", supported_extensions=[".md"])
|
||||
events = [e async for e in src.discover()]
|
||||
assert len(events) == 1
|
||||
assert events[0].kind is SourceEventKind.UPSERT
|
||||
|
|
|
|||
|
|
@ -384,3 +384,53 @@ async def test_discover_url_decodes_href_path():
|
|||
assert [e.uri for e in events] == [
|
||||
"https://nc.example.com/dav/my docs/Hello World.md"
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_emits_unchanged_for_known_uri_without_revision():
|
||||
"""A WebDAV entry with no ETag or Last-Modified should not cause
|
||||
re-ingestion every sweep once the URI has been ingested."""
|
||||
multistatus = _multistatus(
|
||||
{"href": "/dav/", "collection": True},
|
||||
{"href": "/dav/norev.md", "content_type": "text/markdown"},
|
||||
)
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(207, content=multistatus)
|
||||
|
||||
src = WebDAVSource(
|
||||
source_id="nc",
|
||||
base_url="https://nc.example.com/dav/",
|
||||
transport=_transport(handler),
|
||||
)
|
||||
events = [
|
||||
e
|
||||
async for e in src.discover(
|
||||
known_uris={"https://nc.example.com/dav/norev.md"}
|
||||
)
|
||||
]
|
||||
non_delete = [e for e in events if e.kind is not SourceEventKind.DELETE]
|
||||
assert len(non_delete) == 1
|
||||
assert non_delete[0].kind is SourceEventKind.UNCHANGED
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_discover_emits_upsert_for_unknown_uri_without_revision():
|
||||
"""A brand-new WebDAV entry with no revision should UPSERT on first sight."""
|
||||
multistatus = _multistatus(
|
||||
{"href": "/dav/", "collection": True},
|
||||
{"href": "/dav/new.md", "content_type": "text/markdown"},
|
||||
)
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(207, content=multistatus)
|
||||
|
||||
src = WebDAVSource(
|
||||
source_id="nc",
|
||||
base_url="https://nc.example.com/dav/",
|
||||
transport=_transport(handler),
|
||||
)
|
||||
events = [e async for e in src.discover()]
|
||||
non_delete = [e for e in events if e.kind is not SourceEventKind.DELETE]
|
||||
assert len(non_delete) == 1
|
||||
assert non_delete[0].kind is SourceEventKind.UPSERT
|
||||
|
|
|
|||
Loading…
Reference in a new issue